tessl/pypi-python-engineio

Engine.IO server and client for Python providing real-time bidirectional communication

—

Pending

Overview

Eval results

Files

Synchronous Client

Name: tessl/pypi-python-engineio
Author: tessl

Engine.IO client for connecting to servers from synchronous Python applications. Provides automatic reconnection, transport fallback support, and event-driven communication patterns.

Capabilities

Client Initialization

Create and configure a synchronous Engine.IO client with customization options for timeouts, SSL verification, and transport protocols.

class Client:
    def __init__(
        self,
        logger=False,
        json=None,
        request_timeout=5,
        http_session=None,
        ssl_verify=True,
        handle_sigint=True,
        websocket_extra_options=None,
        timestamp_requests=True
    ):
        """
        Initialize Engine.IO client.

        Args:
            logger (bool|Logger): Logging configuration, default False
            json (module): Alternative JSON module
            request_timeout (int): Request timeout in seconds, default 5
            http_session (requests.Session): HTTP session object for requests
            ssl_verify (bool): Verify SSL certificates, default True
            handle_sigint (bool): Handle SIGINT automatically, default True
            websocket_extra_options (dict): Extra WebSocket client options
            timestamp_requests (bool): Add timestamps to requests, default True
        """

Connection Management

Establish connections to Engine.IO servers with support for custom headers, transport selection, and endpoint configuration.

def connect(self, url, headers=None, transports=None, engineio_path='engine.io'):
    """
    Connect to an Engine.IO server.

    Args:
        url (str): Server URL (e.g., 'http://localhost:5000')
        headers (dict, optional): Additional HTTP headers for connection
        transports (list, optional): Allowed transports ['polling', 'websocket']
        engineio_path (str): Engine.IO endpoint path, default 'engine.io'

    Raises:
        ConnectionError: If connection fails
        ValueError: If URL is invalid or transports unsupported
    """

def wait(self):
    """
    Wait until the connection ends.

    This method blocks the current thread until the client disconnects
    or the connection is lost.
    """

def disconnect(self, abort=False, reason=None):
    """
    Disconnect from the server.

    Args:
        abort (bool): Abort connection immediately without graceful shutdown
        reason (str, optional): Reason for disconnection
    """

Message Communication

Send messages to the server and receive responses through the event system.

def send(self, data):
    """
    Send a message to the server.

    Args:
        data (any): Message data to send (will be JSON-serialized)

    Raises:
        SocketIsClosedError: If not connected to server
    """

Event Handling

def on(self, event, handler=None):
    """
    Register an event handler.

    Args:
        event (str): Event name ('connect', 'message', 'disconnect')
        handler (callable, optional): Event handler function

    Returns:
        callable: Decorator function if handler not provided
    """

Usage examples:

# Decorator syntax
@client.on('connect')
def on_connect():
    print('Connected to server')
    client.send('Hello Server!')

@client.on('message')
def on_message(data):
    print(f'Received from server: {data}')

@client.on('disconnect')
def on_disconnect():
    print('Disconnected from server')

# Method syntax
def handle_message(data):
    print(f'Message: {data}')

client.on('message', handle_message)

Transport Information

Query the current transport being used for the connection.

def transport(self):
    """
    Return the current transport name.

    Returns:
        str: Transport name ('polling' or 'websocket')

    Raises:
        ValueError: If not connected
    """

Background Tasks

Manage background tasks using the appropriate threading model.

def start_background_task(self, target, *args, **kwargs):
    """
    Start a background task.

    Args:
        target (callable): Task function to run
        *args: Arguments for the task function
        **kwargs: Keyword arguments for the task function

    Returns:
        Thread: Thread object for the background task
    """

def sleep(self, seconds=0):
    """
    Sleep for the requested time.

    Args:
        seconds (float): Sleep duration in seconds
    """

Utility Methods

Utility functions for queue management, events, and threading primitives.

def create_queue(self, *args, **kwargs):
    """
    Create a queue object appropriate for threading.

    Returns:
        queue.Queue: Thread-safe queue object
    """

def get_queue_empty_exception(self):
    """
    Return the queue empty exception for threading.

    Returns:
        queue.Empty: Queue empty exception class
    """

def create_event(self, *args, **kwargs):
    """
    Create an event object appropriate for threading.

    Returns:
        threading.Event: Threading event object
    """

Client Lifecycle Events

The client emits the following events during server interactions:

connect: Fired when connection to server is established
- Handler signature: () -> None
- Called after successful connection handshake
message: Fired when a message is received from the server
- Handler signature: (data: any) -> None
- data: Message data received from server
disconnect: Fired when disconnected from the server
- Handler signature: () -> None
- Called for both graceful and unexpected disconnections

Connection Examples

Basic Connection

import engineio

client = engineio.Client()

@client.on('connect')
def on_connect():
    print('Connected!')
    client.send('Hello from client')

@client.on('message')
def on_message(data):
    print(f'Server says: {data}')

client.connect('http://localhost:5000')
client.wait()

Connection with Custom Headers

import engineio

client = engineio.Client()

headers = {
    'Authorization': 'Bearer token123',
    'User-Agent': 'MyApp/1.0'
}

client.connect('https://api.example.com', headers=headers)

Connection with SSL Options

import engineio
import requests

# Create custom HTTP session with SSL settings
session = requests.Session()
session.verify = '/path/to/ca-bundle.crt'

client = engineio.Client(
    http_session=session,
    ssl_verify=True,
    request_timeout=10
)

client.connect('https://secure-server.com')

Transport Selection

import engineio

# Only use WebSocket transport
client = engineio.Client()
client.connect('ws://localhost:5000', transports=['websocket'])

# Only use polling transport
client.connect('http://localhost:5000', transports=['polling'])

Disconnection Reasons

class Client.reason:
    CLIENT_DISCONNECT = 'client disconnect'
    SERVER_DISCONNECT = 'server disconnect'
    TRANSPORT_ERROR = 'transport error'

Configuration Constants

# Valid event names
event_names = ['connect', 'disconnect', 'message']

Error Handling

The client may raise the following exceptions during operation:

ConnectionError: When connection to server fails or is lost
SocketIsClosedError: When attempting to send while not connected
ValueError: When invalid parameters are provided
TimeoutError: When requests exceed the configured timeout

Advanced Usage

Automatic Reconnection

import engineio
import time

client = engineio.Client()

@client.on('connect')
def on_connect():
    print('Connected to server')

@client.on('disconnect')
def on_disconnect():
    print('Disconnected, attempting to reconnect...')
    time.sleep(5)
    try:
        client.connect('http://localhost:5000')
    except Exception as e:
        print(f'Reconnection failed: {e}')

client.connect('http://localhost:5000')
client.wait()

Custom JSON Serialization

import engineio
import ujson  # Ultra-fast JSON library

client = engineio.Client(json=ujson)
client.connect('http://localhost:5000')

Install with Tessl CLI

npx tessl i tessl/pypi-python-engineio

docs