or run

npx @tessl/cli init
Log in

Version

Tile

Overview

Evals

Files

docs

clients.mdexceptions.mdindex.mdintegration.mdmanagers.mdnamespaces.mdservers.md
tile.json

tessl/pypi-python-socketio

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

Workspace
tessl
Visibility
Public
Created
Last updated
Describes
pypipkg:pypi/python-socketio@5.13.x

To install, run

npx @tessl/cli install tessl/pypi-python-socketio@5.13.0

index.mddocs/

Python Socket.IO

A comprehensive Python implementation of the Socket.IO realtime communication protocol, providing both synchronous and asynchronous clients and servers. Socket.IO enables bidirectional real-time communication between web clients and servers through WebSocket connections with fallback support for polling transport.

Package Information

  • Package Name: python-socketio
  • Language: Python (>=3.8)
  • Installation: pip install python-socketio
  • Version: 5.13.0

Core Imports

import socketio

Common specific imports:

from socketio import Server, AsyncServer, Client, AsyncClient
from socketio import SimpleClient, AsyncSimpleClient
from socketio import Namespace, AsyncNamespace
from socketio import WSGIApp, ASGIApp

Basic Usage

Simple Server

import socketio

# Create a Socket.IO server
sio = socketio.Server()

@sio.event
def connect(sid, environ):
    print(f'Client {sid} connected')

@sio.event
def disconnect(sid):
    print(f'Client {sid} disconnected')

@sio.event
def message(sid, data):
    # Echo the message back to the client
    sio.emit('response', {'echo': data}, room=sid)

# WSGI application
app = socketio.WSGIApp(sio)

Simple Client

import socketio

# Create a Socket.IO client
sio = socketio.Client()

@sio.event
def connect():
    print('Connected to server')
    sio.emit('message', {'hello': 'world'})

@sio.event
def response(data):
    print(f'Received: {data}')

# Connect to server
sio.connect('http://localhost:5000')

Asynchronous Server

import socketio

# Create an async Socket.IO server
sio = socketio.AsyncServer()

@sio.event
async def connect(sid, environ):
    print(f'Client {sid} connected')

@sio.event
async def message(sid, data):
    await sio.emit('response', {'echo': data}, room=sid)

# ASGI application
app = socketio.ASGIApp(sio)

Architecture

The python-socketio library provides a hierarchical architecture supporting both synchronous and asynchronous operation patterns:

  • Client/Server Classes: Core communication endpoints (sync and async variants)
  • Manager Classes: Handle client connections and message routing, with support for scaling across multiple processes using Redis, Kafka, RabbitMQ, and ZeroMQ
  • Namespace Classes: Organize communication into logical groups with class-based event handlers
  • Middleware Classes: Integration adapters for web frameworks (WSGI, ASGI, Tornado)
  • Transport Support: Automatic WebSocket with polling fallback through Engine.IO integration

This design enables everything from simple single-process applications to horizontally-scaled real-time systems while maintaining compatibility with the JavaScript Socket.IO ecosystem.

Capabilities

Client Classes

Socket.IO client implementations for connecting to servers, supporting both simple blocking operations and full-featured event-driven communication with automatic reconnection and namespace support.

class SimpleClient:
    def connect(self, url, **kwargs): ...
    def emit(self, event, data=None): ...
    def call(self, event, data=None, timeout=60): ...
    def receive(self, timeout=None): ...
    def disconnect(self): ...

class Client:
    def connect(self, url, **kwargs): ...
    def emit(self, event, data=None, namespace=None, callback=None): ...
    def call(self, event, data=None, namespace=None, timeout=60): ...
    def wait(self): ...
    def disconnect(self): ...

class AsyncSimpleClient:
    async def connect(self, url, **kwargs): ...
    async def emit(self, event, data=None): ...
    async def call(self, event, data=None, timeout=60): ...
    async def receive(self, timeout=None): ...
    async def disconnect(self): ...

class AsyncClient:
    async def connect(self, url, **kwargs): ...
    async def emit(self, event, data=None, namespace=None, callback=None): ...
    async def call(self, event, data=None, namespace=None, timeout=60): ...
    async def wait(self): ...
    async def disconnect(self): ...

Client Classes

Server Classes

Socket.IO server implementations for handling multiple client connections, supporting event-driven architecture, room management, session handling, and client manager integration for horizontal scaling.

class Server:
    def emit(self, event, data=None, to=None, room=None, skip_sid=None, namespace=None, callback=None): ...
    def send(self, data, to=None, room=None, skip_sid=None, namespace=None, callback=None): ...
    def call(self, event, data=None, to=None, sid=None, namespace=None, timeout=60): ...
    def enter_room(self, sid, room, namespace=None): ...
    def leave_room(self, sid, room, namespace=None): ...
    def close_room(self, room, namespace=None): ...
    def get_session(self, sid, namespace=None): ...
    def save_session(self, sid, session, namespace=None): ...
    def disconnect(self, sid, namespace=None): ...

class AsyncServer:
    async def emit(self, event, data=None, to=None, room=None, skip_sid=None, namespace=None, callback=None): ...
    async def send(self, data, to=None, room=None, skip_sid=None, namespace=None, callback=None): ...
    async def call(self, event, data=None, to=None, sid=None, namespace=None, timeout=60): ...
    async def enter_room(self, sid, room, namespace=None): ...
    async def leave_room(self, sid, room, namespace=None): ...
    async def close_room(self, room, namespace=None): ...
    async def get_session(self, sid, namespace=None): ...
    async def save_session(self, sid, session, namespace=None): ...
    async def disconnect(self, sid, namespace=None): ...

Server Classes

Namespace Classes

Class-based event handlers that organize Socket.IO communication into logical namespaces, supporting inheritance and method-based event handling for both client and server scenarios.

class Namespace:
    def trigger_event(self, event, *args): ...
    def emit(self, event, data=None, to=None, room=None, skip_sid=None, namespace=None, callback=None): ...
    def send(self, data, to=None, room=None, skip_sid=None, namespace=None, callback=None): ...
    def call(self, event, data=None, to=None, sid=None, namespace=None, timeout=60): ...
    def enter_room(self, sid, room, namespace=None): ...
    def leave_room(self, sid, room, namespace=None): ...
    def disconnect(self, sid, namespace=None): ...

class ClientNamespace:
    def trigger_event(self, event, *args): ...
    def emit(self, event, data=None, namespace=None, callback=None): ...
    def send(self, data, namespace=None, callback=None): ...
    def call(self, event, data=None, namespace=None, timeout=60): ...
    def disconnect(self): ...

class AsyncNamespace:
    async def trigger_event(self, event, *args): ...
    async def emit(self, event, data=None, to=None, room=None, skip_sid=None, namespace=None, callback=None): ...
    async def send(self, data, to=None, room=None, skip_sid=None, namespace=None, callback=None): ...
    async def call(self, event, data=None, to=None, sid=None, namespace=None, timeout=60): ...
    async def enter_room(self, sid, room, namespace=None): ...
    async def leave_room(self, sid, room, namespace=None): ...
    async def disconnect(self, sid, namespace=None): ...

class AsyncClientNamespace:
    async def trigger_event(self, event, *args): ...
    async def emit(self, event, data=None, namespace=None, callback=None): ...
    async def send(self, data, namespace=None, callback=None): ...
    async def call(self, event, data=None, namespace=None, timeout=60): ...
    async def disconnect(self): ...

Namespace Classes

Manager Classes

Client connection managers that handle message routing, room membership, and horizontal scaling through pub/sub messaging systems like Redis, Kafka, RabbitMQ, and ZeroMQ.

class Manager:
    def emit(self, event, data, namespace, room=None, skip_sid=None, callback=None, **kwargs): ...
    def can_disconnect(self, sid, namespace): ...
    def disconnect(self, sid, namespace=None): ...
    def enter_room(self, sid, namespace, room, eio_sid=None): ...
    def leave_room(self, sid, namespace, room): ...
    def close_room(self, room, namespace): ...

class PubSubManager(Manager):
    def __init__(self, channel='socketio', write_only=False, logger=None): ...

class RedisManager(PubSubManager):
    def __init__(self, url='redis://localhost:6379/0', channel='socketio', write_only=False, logger=None, redis_options=None): ...

class KafkaManager(PubSubManager):
    def __init__(self, url='kafka://localhost:9092', channel='socketio', write_only=False, logger=None): ...

Manager Classes

Web Framework Integration

Middleware classes for integrating Socket.IO servers with popular Python web frameworks, supporting both WSGI and ASGI patterns with static file serving and flexible routing.

class WSGIApp:
    def __init__(self, socketio_app, wsgi_app=None, static_files=None, socketio_path='socket.io'): ...

class ASGIApp:
    def __init__(self, socketio_server, other_asgi_app=None, static_files=None, socketio_path='socket.io', on_startup=None, on_shutdown=None): ...

def get_tornado_handler(socketio_server):
    """Get Tornado request handler for Socket.IO server.
    
    Returns:
        Tornado handler class for Socket.IO integration
    """

Web Framework Integration

Exception Handling

Comprehensive exception hierarchy for handling Socket.IO-specific errors including connection failures, timeouts, namespace issues, and disconnection scenarios.

class SocketIOError(Exception): ...

class ConnectionError(SocketIOError): ...

class ConnectionRefusedError(ConnectionError):
    def __init__(self, *args): ...

class TimeoutError(SocketIOError): ...

class BadNamespaceError(SocketIOError): ...

class DisconnectedError(SocketIOError): ...

Exception Handling