tessl/pypi-python-socketio
Socket.IO server and client for Python providing real-time bidirectional communication
—
Pending
servers.mddocs/
Server Classes
Socket.IO server implementations for handling multiple client connections with support for event-driven architecture, room management, session handling, and horizontal scaling through client managers.
Capabilities
Server
A synchronous Socket.IO server that handles multiple client connections, provides event-driven communication, room management, session storage, and integrates with various client managers for scaling.
class Server:
"""
Synchronous Socket.IO server with full features.
Inherits from: BaseServer
Attributes:
manager: Client manager instance for handling connections
logger: Logger instance for server events
eio: Engine.IO server instance
"""
def __init__(self, client_manager=None, logger=False, serializer='default', json=None,
async_handlers=False, always_connect=False, namespaces=None, **kwargs):
"""
Initialize the server.
Args:
client_manager: Client manager instance (defaults to Manager())
logger (bool or Logger): Enable logging or provide custom logger
serializer (str): Message serializer ('default', 'pickle', 'msgpack', 'cbor')
json (module): Custom JSON module
async_handlers (bool): Allow async event handlers in sync server
always_connect (bool): Accept all connections without authentication
namespaces (list): Allowed namespaces
**kwargs: Additional Engine.IO server parameters
"""
def emit(self, event, data=None, to=None, room=None, skip_sid=None, namespace=None, callback=None):
"""
Emit an event to one or more connected clients.
Args:
event (str): Event name
data: Event data to send
to (str or list): Target client session ID(s)
room (str or list): Target room name(s)
skip_sid (str): Skip this client session ID
namespace (str): Target namespace
callback (callable): Callback function for response
Raises:
DisconnectedError: Target client not connected
"""
def send(self, data, to=None, room=None, skip_sid=None, namespace=None, callback=None):
"""
Send a message event to one or more connected clients.
Args:
data: Message data to send
to (str or list): Target client session ID(s)
room (str or list): Target room name(s)
skip_sid (str): Skip this client session ID
namespace (str): Target namespace
callback (callable): Callback function for response
Raises:
DisconnectedError: Target client not connected
"""
def call(self, event, data=None, to=None, sid=None, namespace=None, timeout=60):
"""
Emit an event and wait for a response from a specific client.
Args:
event (str): Event name
data: Event data to send
to (str): Target client session ID (deprecated, use sid)
sid (str): Target client session ID
namespace (str): Target namespace
timeout (int): Response timeout in seconds
Returns:
Response data from client
Raises:
TimeoutError: No response received within timeout
DisconnectedError: Target client not connected
"""
def enter_room(self, sid, room, namespace=None):
"""
Add a client to a room.
Args:
sid (str): Client session ID
room (str): Room name
namespace (str): Target namespace
Raises:
DisconnectedError: Client not connected
"""
def leave_room(self, sid, room, namespace=None):
"""
Remove a client from a room.
Args:
sid (str): Client session ID
room (str): Room name
namespace (str): Target namespace
Raises:
DisconnectedError: Client not connected
"""
def close_room(self, room, namespace=None):
"""
Remove all clients from a room and delete the room.
Args:
room (str): Room name
namespace (str): Target namespace
"""
def rooms(self, sid, namespace=None):
"""
Get the list of rooms a client has joined.
Args:
sid (str): Client session ID
namespace (str): Target namespace
Returns:
list: Room names the client has joined
Raises:
DisconnectedError: Client not connected
"""
def get_session(self, sid, namespace=None):
"""
Get the session data for a client.
Args:
sid (str): Client session ID
namespace (str): Target namespace
Returns:
dict: Session data
Raises:
DisconnectedError: Client not connected
"""
def save_session(self, sid, session, namespace=None):
"""
Save session data for a client.
Args:
sid (str): Client session ID
session (dict): Session data to save
namespace (str): Target namespace
Raises:
DisconnectedError: Client not connected
"""
def session(self, sid, namespace=None):
"""
Access client session data with context manager.
Args:
sid (str): Client session ID
namespace (str): Target namespace
Returns:
Context manager for session access
Raises:
DisconnectedError: Client not connected
"""
def disconnect(self, sid, namespace=None):
"""
Disconnect a client.
Args:
sid (str): Client session ID
namespace (str): Target namespace
Raises:
DisconnectedError: Client not connected
"""
def start_background_task(self, target, *args, **kwargs):
"""
Start a background task.
Args:
target (callable): Task function
*args: Task arguments
**kwargs: Task keyword arguments
Returns:
Task handle
"""
def sleep(self, seconds):
"""
Sleep for the given number of seconds.
Args:
seconds (float): Sleep duration
"""
def on(self, event, handler=None, namespace=None):
"""
Register an event handler.
Args:
event (str): Event name
handler (callable): Event handler function
namespace (str): Target namespace
Returns:
Decorator function if handler not provided
"""
def event(self, event=None, handler=None, namespace=None):
"""
Decorator to register an event handler.
Args:
event (str, optional): Event name (defaults to function name)
handler (callable): Event handler function
namespace (str): Target namespace
Returns:
Decorator function
"""Usage Example
import socketio
# Create server with Redis manager for scaling
redis_manager = socketio.RedisManager('redis://localhost:6379')
sio = socketio.Server(client_manager=redis_manager, logger=True)
# Event handlers
@sio.event
def connect(sid, environ):
print(f'Client {sid} connected')
# Add to default room
sio.enter_room(sid, 'lobby')
sio.emit('welcome', {'message': 'Welcome to the server!'}, room=sid)
@sio.event
def disconnect(sid):
print(f'Client {sid} disconnected')
@sio.event
def join_room(sid, data):
room = data['room']
sio.enter_room(sid, room)
sio.emit('status', {'message': f'Joined room {room}'}, room=sid)
sio.emit('user_joined', {'sid': sid}, room=room, skip_sid=sid)
@sio.event
def leave_room(sid, data):
room = data['room']
sio.leave_room(sid, room)
sio.emit('user_left', {'sid': sid}, room=room)
@sio.event
def chat_message(sid, data):
# Get user session data
with sio.session(sid) as session:
username = session.get('username', 'Anonymous')
room = data.get('room', 'lobby')
message = {
'username': username,
'message': data['message'],
'timestamp': time.time()
}
# Broadcast to room
sio.emit('message', message, room=room)
@sio.event
def set_username(sid, data):
# Save username in session
with sio.session(sid) as session:
session['username'] = data['username']
sio.emit('username_set', {'success': True}, room=sid)
@sio.event
def get_room_info(sid, data):
room = data['room']
# Use call to get response
try:
info = sio.call('room_info_request', {'room': room}, sid=sid, timeout=5)
return {'info': info}
except socketio.TimeoutError:
return {'error': 'Timeout getting room info'}
# Create WSGI app
app = socketio.WSGIApp(sio)AsyncServer
Asynchronous Socket.IO server that provides all the capabilities of the synchronous Server but uses coroutines for non-blocking operation in asyncio applications.
class AsyncServer:
"""
Asynchronous Socket.IO server for asyncio.
Inherits from: BaseServer
Attributes:
manager: Async client manager instance
logger: Logger instance for server events
eio: Engine.IO async server instance
"""
def __init__(self, client_manager=None, logger=False, serializer='default', json=None,
always_connect=False, namespaces=None, **kwargs):
"""
Initialize the async server.
Args:
client_manager: Async client manager instance (defaults to AsyncManager())
logger (bool or Logger): Enable logging or provide custom logger
serializer (str): Message serializer ('default', 'pickle', 'msgpack', 'cbor')
json (module): Custom JSON module
always_connect (bool): Accept all connections without authentication
namespaces (list): Allowed namespaces
**kwargs: Additional Engine.IO server parameters
"""
async def emit(self, event, data=None, to=None, room=None, skip_sid=None, namespace=None, callback=None):
"""
Emit an event to one or more connected clients.
Args:
event (str): Event name
data: Event data to send
to (str or list): Target client session ID(s)
room (str or list): Target room name(s)
skip_sid (str): Skip this client session ID
namespace (str): Target namespace
callback (coroutine): Async callback function for response
Raises:
DisconnectedError: Target client not connected
"""
async def send(self, data, to=None, room=None, skip_sid=None, namespace=None, callback=None):
"""
Send a message event to one or more connected clients.
Args:
data: Message data to send
to (str or list): Target client session ID(s)
room (str or list): Target room name(s)
skip_sid (str): Skip this client session ID
namespace (str): Target namespace
callback (coroutine): Async callback function for response
Raises:
DisconnectedError: Target client not connected
"""
async def call(self, event, data=None, to=None, sid=None, namespace=None, timeout=60):
"""
Emit an event and wait for a response from a specific client.
Args:
event (str): Event name
data: Event data to send
to (str): Target client session ID (deprecated, use sid)
sid (str): Target client session ID
namespace (str): Target namespace
timeout (int): Response timeout in seconds
Returns:
Response data from client
Raises:
TimeoutError: No response received within timeout
DisconnectedError: Target client not connected
"""
async def enter_room(self, sid, room, namespace=None):
"""
Add a client to a room.
Args:
sid (str): Client session ID
room (str): Room name
namespace (str): Target namespace
Raises:
DisconnectedError: Client not connected
"""
async def leave_room(self, sid, room, namespace=None):
"""
Remove a client from a room.
Args:
sid (str): Client session ID
room (str): Room name
namespace (str): Target namespace
Raises:
DisconnectedError: Client not connected
"""
async def close_room(self, room, namespace=None):
"""
Remove all clients from a room and delete the room.
Args:
room (str): Room name
namespace (str): Target namespace
"""
async def rooms(self, sid, namespace=None):
"""
Get the list of rooms a client has joined.
Args:
sid (str): Client session ID
namespace (str): Target namespace
Returns:
list: Room names the client has joined
Raises:
DisconnectedError: Client not connected
"""
async def get_session(self, sid, namespace=None):
"""
Get the session data for a client.
Args:
sid (str): Client session ID
namespace (str): Target namespace
Returns:
dict: Session data
Raises:
DisconnectedError: Client not connected
"""
async def save_session(self, sid, session, namespace=None):
"""
Save session data for a client.
Args:
sid (str): Client session ID
session (dict): Session data to save
namespace (str): Target namespace
Raises:
DisconnectedError: Client not connected
"""
def session(self, sid, namespace=None):
"""
Access client session data with async context manager.
Args:
sid (str): Client session ID
namespace (str): Target namespace
Returns:
Async context manager for session access
Raises:
DisconnectedError: Client not connected
"""
async def disconnect(self, sid, namespace=None):
"""
Disconnect a client.
Args:
sid (str): Client session ID
namespace (str): Target namespace
Raises:
DisconnectedError: Client not connected
"""
def start_background_task(self, target, *args, **kwargs):
"""
Start a background task.
Args:
target (coroutine): Async task function
*args: Task arguments
**kwargs: Task keyword arguments
Returns:
asyncio.Task: Task handle
"""
async def sleep(self, seconds):
"""
Sleep for the given number of seconds.
Args:
seconds (float): Sleep duration
"""
def on(self, event, handler=None, namespace=None):
"""
Register an async event handler.
Args:
event (str): Event name
handler (coroutine): Async event handler function
namespace (str): Target namespace
Returns:
Decorator function if handler not provided
"""
def event(self, event=None, handler=None, namespace=None):
"""
Decorator to register an async event handler.
Args:
event (str, optional): Event name (defaults to function name)
handler (coroutine): Async event handler function
namespace (str): Target namespace
Returns:
Decorator function
"""Usage Example
import asyncio
import socketio
# Create async server with Redis manager
redis_manager = socketio.AsyncRedisManager('redis://localhost:6379')
sio = socketio.AsyncServer(client_manager=redis_manager, logger=True)
# Async event handlers
@sio.event
async def connect(sid, environ):
print(f'Client {sid} connected')
await sio.enter_room(sid, 'lobby')
await sio.emit('welcome', {'message': 'Welcome!'}, room=sid)
@sio.event
async def disconnect(sid):
print(f'Client {sid} disconnected')
@sio.event
async def join_room(sid, data):
room = data['room']
await sio.enter_room(sid, room)
await sio.emit('status', {'message': f'Joined {room}'}, room=sid)
await sio.emit('user_joined', {'sid': sid}, room=room, skip_sid=sid)
@sio.event
async def chat_message(sid, data):
# Access session asynchronously
async with sio.session(sid) as session:
username = session.get('username', 'Anonymous')
room = data.get('room', 'lobby')
message = {
'username': username,
'message': data['message'],
'timestamp': time.time()
}
# Broadcast to room
await sio.emit('message', message, room=room)
@sio.event
async def set_username(sid, data):
async with sio.session(sid) as session:
session['username'] = data['username']
await sio.emit('username_set', {'success': True}, room=sid)
@sio.event
async def get_users_in_room(sid, data):
room = data['room']
# Simulate async database lookup
await asyncio.sleep(0.1)
users = await get_room_users(room) # Your async function
return {'users': users}
async def get_room_users(room):
# Your async implementation
return ['user1', 'user2', 'user3']
# Create ASGI app
app = socketio.ASGIApp(sio)
# Run with an ASGI server like uvicorn
# uvicorn main:app --host 0.0.0.0 --port 5000Event Handler Patterns
Both Server and AsyncServer support multiple patterns for event handlers:
Function-based Handlers
# Synchronous server
sio = socketio.Server()
@sio.event
def connect(sid, environ):
print(f'Client {sid} connected')
# With custom event name
@sio.on('custom_event')
def handle_custom(sid, data):
print(f'Custom event from {sid}: {data}')
# Async server
async_sio = socketio.AsyncServer()
@async_sio.event
async def connect(sid, environ):
print(f'Client {sid} connected')
await async_sio.emit('welcome', room=sid)Class-based Handlers (Namespaces)
# Using namespace classes (covered in detail in namespaces.md)
class ChatNamespace(socketio.Namespace):
def on_connect(self, sid, environ):
print(f'Client {sid} connected to chat')
def on_message(self, sid, data):
self.emit('response', data, room=sid)
sio.register_namespace(ChatNamespace('/chat'))Session Management
Both servers provide comprehensive session management:
# Synchronous session management
@sio.event
def save_user_data(sid, data):
# Direct session access
sio.save_session(sid, {'user_id': data['user_id'], 'preferences': data['prefs']})
# Context manager (recommended)
with sio.session(sid) as session:
session['last_activity'] = time.time()
session['status'] = 'active'
# Asynchronous session management
@async_sio.event
async def save_user_data(sid, data):
# Direct session access
await async_sio.save_session(sid, {'user_id': data['user_id']})
# Async context manager (recommended)
async with async_sio.session(sid) as session:
session['last_activity'] = time.time()
session['status'] = 'active'Install with Tessl CLI
npx tessl i tessl/pypi-python-socketio