tessl/pypi-python-socketio
Socket.IO server and client for Python providing real-time bidirectional communication
—
Pending
namespaces.mddocs/
Namespace Classes
Class-based event handlers that organize Socket.IO communication into logical namespaces. Namespaces enable modular application design by separating different functional areas and providing inheritance-based event handling patterns.
Capabilities
Namespace
Server-side base class for creating class-based event handlers in synchronous applications. Provides method-based event handling and access to server functionality through inheritance.
class Namespace:
"""
Base class for server-side class-based namespaces (synchronous).
Inherits from: BaseServerNamespace
Attributes:
namespace (str): Namespace path (default: '/')
server: Reference to the Socket.IO server instance
"""
def __init__(self, namespace='/'):
"""
Initialize the namespace.
Args:
namespace (str): Namespace path (default: '/')
"""
def trigger_event(self, event, *args):
"""
Dispatch events to handler methods.
Args:
event (str): Event name
*args: Event arguments
Returns:
Handler method return value
"""
def emit(self, event, data=None, to=None, room=None, skip_sid=None, namespace=None, callback=None):
"""
Emit an event to 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 (defaults to this namespace)
callback (callable): Callback function for response
"""
def send(self, data, to=None, room=None, skip_sid=None, namespace=None, callback=None):
"""
Send a message event to 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 (defaults to this namespace)
callback (callable): Callback function for response
"""
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 (defaults to this namespace)
timeout (int): Response timeout in seconds
Returns:
Response data from client
Raises:
TimeoutError: No response received within timeout
"""
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 (defaults to this namespace)
"""
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 (defaults to this namespace)
"""
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 (defaults to this 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 (defaults to this namespace)
Returns:
list: Room names the client has joined
"""
def get_session(self, sid, namespace=None):
"""
Get the session data for a client.
Args:
sid (str): Client session ID
namespace (str): Target namespace (defaults to this namespace)
Returns:
dict: Session data
"""
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 (defaults to this namespace)
"""
def session(self, sid, namespace=None):
"""
Access client session data with context manager.
Args:
sid (str): Client session ID
namespace (str): Target namespace (defaults to this namespace)
Returns:
Context manager for session access
"""
def disconnect(self, sid, namespace=None):
"""
Disconnect a client.
Args:
sid (str): Client session ID
namespace (str): Target namespace (defaults to this namespace)
"""
# Event handler methods (override in subclass)
def on_connect(self, sid, environ):
"""
Called when a client connects.
Args:
sid (str): Client session ID
environ (dict): WSGI environ dictionary
"""
pass
def on_disconnect(self, sid):
"""
Called when a client disconnects.
Args:
sid (str): Client session ID
"""
pass
def on_message(self, sid, data):
"""
Called when a client sends a message event.
Args:
sid (str): Client session ID
data: Message data
"""
passUsage Example
import socketio
class ChatNamespace(socketio.Namespace):
def on_connect(self, sid, environ):
print(f'Client {sid} connected to chat namespace')
self.enter_room(sid, 'lobby')
self.emit('welcome', {'message': 'Welcome to chat!'}, room=sid)
def on_disconnect(self, sid):
print(f'Client {sid} disconnected from chat')
def on_join_room(self, sid, data):
room = data['room']
self.enter_room(sid, room)
self.emit('status', {'message': f'Joined {room}'}, room=sid)
self.emit('user_joined', {'sid': sid}, room=room, skip_sid=sid)
def on_leave_room(self, sid, data):
room = data['room']
self.leave_room(sid, room)
self.emit('user_left', {'sid': sid}, room=room)
def on_chat_message(self, sid, data):
# Get user info from session
with self.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
self.emit('message', message, room=room)
def on_set_username(self, sid, data):
with self.session(sid) as session:
session['username'] = data['username']
self.emit('username_set', {'success': True}, room=sid)
class GameNamespace(socketio.Namespace):
def __init__(self):
super().__init__('/game')
self.games = {}
def on_connect(self, sid, environ):
print(f'Player {sid} connected to game namespace')
def on_create_game(self, sid, data):
game_id = data['game_id']
self.games[game_id] = {'players': [sid], 'status': 'waiting'}
self.enter_room(sid, game_id)
self.emit('game_created', {'game_id': game_id}, room=sid)
def on_join_game(self, sid, data):
game_id = data['game_id']
if game_id in self.games:
self.games[game_id]['players'].append(sid)
self.enter_room(sid, game_id)
self.emit('player_joined', {'sid': sid}, room=game_id)
# Start game if enough players
if len(self.games[game_id]['players']) >= 2:
self.games[game_id]['status'] = 'playing'
self.emit('game_started', room=game_id)
# Register namespaces with server
sio = socketio.Server()
sio.register_namespace(ChatNamespace('/chat'))
sio.register_namespace(GameNamespace())AsyncNamespace
Asynchronous version of the Namespace class for server-side class-based event handlers in asyncio applications.
class AsyncNamespace:
"""
Base class for asyncio server-side class-based namespaces.
Inherits from: BaseServerNamespace
Attributes:
namespace (str): Namespace path (default: '/')
server: Reference to the AsyncServer instance
"""
def __init__(self, namespace='/'):
"""
Initialize the async namespace.
Args:
namespace (str): Namespace path (default: '/')
"""
async def trigger_event(self, event, *args):
"""
Dispatch events to async handler methods.
Args:
event (str): Event name
*args: Event arguments
Returns:
Handler method return value
"""
async def emit(self, event, data=None, to=None, room=None, skip_sid=None, namespace=None, callback=None):
"""
Emit an event to 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 (defaults to this namespace)
callback (coroutine): Async callback function for response
"""
async def send(self, data, to=None, room=None, skip_sid=None, namespace=None, callback=None):
"""
Send a message event to 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 (defaults to this namespace)
callback (coroutine): Async callback function for response
"""
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 (defaults to this namespace)
timeout (int): Response timeout in seconds
Returns:
Response data from client
Raises:
TimeoutError: No response received within timeout
"""
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 (defaults to this namespace)
"""
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 (defaults to this namespace)
"""
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 (defaults to this 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 (defaults to this namespace)
Returns:
list: Room names the client has joined
"""
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 (defaults to this namespace)
Returns:
dict: Session data
"""
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 (defaults to this namespace)
"""
def session(self, sid, namespace=None):
"""
Access client session data with async context manager.
Args:
sid (str): Client session ID
namespace (str): Target namespace (defaults to this namespace)
Returns:
Async context manager for session access
"""
async def disconnect(self, sid, namespace=None):
"""
Disconnect a client.
Args:
sid (str): Client session ID
namespace (str): Target namespace (defaults to this namespace)
"""
# Async event handler methods (override in subclass)
async def on_connect(self, sid, environ):
"""
Called when a client connects.
Args:
sid (str): Client session ID
environ (dict): WSGI environ dictionary
"""
pass
async def on_disconnect(self, sid):
"""
Called when a client disconnects.
Args:
sid (str): Client session ID
"""
pass
async def on_message(self, sid, data):
"""
Called when a client sends a message event.
Args:
sid (str): Client session ID
data: Message data
"""
passUsage Example
import asyncio
import socketio
class AsyncChatNamespace(socketio.AsyncNamespace):
def __init__(self):
super().__init__('/chat')
self.user_cache = {}
async def on_connect(self, sid, environ):
print(f'Client {sid} connected to async chat namespace')
await self.enter_room(sid, 'lobby')
await self.emit('welcome', {'message': 'Welcome!'}, room=sid)
async def on_disconnect(self, sid):
print(f'Client {sid} disconnected from async chat')
# Clean up user cache
if sid in self.user_cache:
del self.user_cache[sid]
async def on_join_room(self, sid, data):
room = data['room']
await self.enter_room(sid, room)
await self.emit('status', {'message': f'Joined {room}'}, room=sid)
await self.emit('user_joined', {'sid': sid}, room=room, skip_sid=sid)
async def on_chat_message(self, sid, data):
# Async session access
async with self.session(sid) as session:
username = session.get('username', 'Anonymous')
# Simulate async processing (e.g., content filtering)
await asyncio.sleep(0.01)
room = data.get('room', 'lobby')
message = {
'username': username,
'message': data['message'],
'timestamp': time.time()
}
# Broadcast to room
await self.emit('message', message, room=room)
async def on_get_user_info(self, sid, data):
# Simulate async database lookup
user_id = data['user_id']
user_info = await self.fetch_user_info(user_id)
return {'user_info': user_info}
async def fetch_user_info(self, user_id):
# Simulate async database call
await asyncio.sleep(0.1)
return {'id': user_id, 'name': f'User {user_id}', 'status': 'online'}
class AsyncGameNamespace(socketio.AsyncNamespace):
def __init__(self):
super().__init__('/game')
self.games = {}
self.game_lock = asyncio.Lock()
async def on_connect(self, sid, environ):
print(f'Player {sid} connected to async game namespace')
async def on_create_game(self, sid, data):
game_id = data['game_id']
async with self.game_lock:
self.games[game_id] = {
'players': [sid],
'status': 'waiting',
'created_at': time.time()
}
await self.enter_room(sid, game_id)
await self.emit('game_created', {'game_id': game_id}, room=sid)
async def on_join_game(self, sid, data):
game_id = data['game_id']
async with self.game_lock:
if game_id in self.games:
self.games[game_id]['players'].append(sid)
await self.enter_room(sid, game_id)
await self.emit('player_joined', {'sid': sid}, room=game_id)
# Start game if enough players
if len(self.games[game_id]['players']) >= 2:
self.games[game_id]['status'] = 'playing'
await self.emit('game_started', room=game_id)
# Start game logic asynchronously
asyncio.create_task(self.run_game(game_id))
async def run_game(self, game_id):
# Simulate async game logic
await asyncio.sleep(1)
await self.emit('game_update', {'status': 'running'}, room=game_id)
# Register namespaces with async server
sio = socketio.AsyncServer()
sio.register_namespace(AsyncChatNamespace())
sio.register_namespace(AsyncGameNamespace())ClientNamespace
Base class for client-side class-based event handlers in synchronous client applications.
class ClientNamespace:
"""
Base class for client-side class-based namespaces (synchronous).
Inherits from: BaseClientNamespace
Attributes:
namespace (str): Namespace path (default: '/')
client: Reference to the Client instance
"""
def __init__(self, namespace='/'):
"""
Initialize the client namespace.
Args:
namespace (str): Namespace path (default: '/')
"""
def trigger_event(self, event, *args):
"""
Dispatch events to handler methods.
Args:
event (str): Event name
*args: Event arguments
Returns:
Handler method return value
"""
def emit(self, event, data=None, namespace=None, callback=None):
"""
Emit an event to the server.
Args:
event (str): Event name
data: Event data to send
namespace (str): Target namespace (defaults to this namespace)
callback (callable): Callback function for response
"""
def send(self, data, namespace=None, callback=None):
"""
Send a message event to the server.
Args:
data: Message data to send
namespace (str): Target namespace (defaults to this namespace)
callback (callable): Callback function for response
"""
def call(self, event, data=None, namespace=None, timeout=60):
"""
Emit an event and wait for a response from the server.
Args:
event (str): Event name
data: Event data to send
namespace (str): Target namespace (defaults to this namespace)
timeout (int): Response timeout in seconds
Returns:
Response data from server
Raises:
TimeoutError: No response received within timeout
"""
def disconnect(self):
"""
Disconnect from the server.
"""
# Event handler methods (override in subclass)
def on_connect(self):
"""Called when connected to the server."""
pass
def on_disconnect(self):
"""Called when disconnected from the server."""
pass
def on_message(self, data):
"""Called when receiving a message event from the server."""
passUsage Example
import socketio
class ChatClientNamespace(socketio.ClientNamespace):
def __init__(self):
super().__init__('/chat')
self.username = None
def on_connect(self):
print('Connected to chat namespace')
self.emit('set_username', {'username': 'Alice'})
def on_disconnect(self):
print('Disconnected from chat namespace')
def on_welcome(self, data):
print(f'Welcome message: {data["message"]}')
def on_message(self, data):
print(f'{data["username"]}: {data["message"]}')
def on_username_set(self, data):
if data['success']:
print('Username set successfully')
self.emit('join_room', {'room': 'general'})
def send_message(self, message, room='general'):
self.emit('chat_message', {
'message': message,
'room': room
})
# Use with client
sio = socketio.Client()
chat_ns = ChatClientNamespace()
sio.register_namespace(chat_ns)
sio.connect('http://localhost:5000')
# Send messages through namespace
chat_ns.send_message('Hello everyone!')
sio.wait()AsyncClientNamespace
Asynchronous version of ClientNamespace for client-side class-based event handlers in asyncio client applications.
class AsyncClientNamespace:
"""
Base class for asyncio client-side class-based namespaces.
Inherits from: BaseClientNamespace
Attributes:
namespace (str): Namespace path (default: '/')
client: Reference to the AsyncClient instance
"""
def __init__(self, namespace='/'):
"""
Initialize the async client namespace.
Args:
namespace (str): Namespace path (default: '/')
"""
async def trigger_event(self, event, *args):
"""
Dispatch events to async handler methods.
Args:
event (str): Event name
*args: Event arguments
Returns:
Handler method return value
"""
async def emit(self, event, data=None, namespace=None, callback=None):
"""
Emit an event to the server.
Args:
event (str): Event name
data: Event data to send
namespace (str): Target namespace (defaults to this namespace)
callback (coroutine): Async callback function for response
"""
async def send(self, data, namespace=None, callback=None):
"""
Send a message event to the server.
Args:
data: Message data to send
namespace (str): Target namespace (defaults to this namespace)
callback (coroutine): Async callback function for response
"""
async def call(self, event, data=None, namespace=None, timeout=60):
"""
Emit an event and wait for a response from the server.
Args:
event (str): Event name
data: Event data to send
namespace (str): Target namespace (defaults to this namespace)
timeout (int): Response timeout in seconds
Returns:
Response data from server
Raises:
TimeoutError: No response received within timeout
"""
async def disconnect(self):
"""
Disconnect from the server.
"""
# Async event handler methods (override in subclass)
async def on_connect(self):
"""Called when connected to the server."""
pass
async def on_disconnect(self):
"""Called when disconnected from the server."""
pass
async def on_message(self, data):
"""Called when receiving a message event from the server."""
passUsage Example
import asyncio
import socketio
class AsyncChatClientNamespace(socketio.AsyncClientNamespace):
def __init__(self):
super().__init__('/chat')
self.message_queue = asyncio.Queue()
async def on_connect(self):
print('Connected to async chat namespace')
await self.emit('set_username', {'username': 'Bob'})
async def on_disconnect(self):
print('Disconnected from async chat namespace')
async def on_welcome(self, data):
print(f'Welcome message: {data["message"]}')
async def on_message(self, data):
print(f'{data["username"]}: {data["message"]}')
# Add to processing queue
await self.message_queue.put(data)
async def on_username_set(self, data):
if data['success']:
print('Username set successfully')
await self.emit('join_room', {'room': 'general'})
async def send_message(self, message, room='general'):
await self.emit('chat_message', {
'message': message,
'room': room
})
async def process_messages(self):
"""Background task to process incoming messages"""
while True:
try:
message = await asyncio.wait_for(
self.message_queue.get(), timeout=1.0
)
# Process message asynchronously
await self.handle_message(message)
except asyncio.TimeoutError:
continue
async def handle_message(self, message):
# Simulate async message processing
await asyncio.sleep(0.1)
print(f'Processed message from {message["username"]}')
async def main():
sio = socketio.AsyncClient()
chat_ns = AsyncChatClientNamespace()
sio.register_namespace(chat_ns)
await sio.connect('http://localhost:5000')
# Start message processing task
process_task = asyncio.create_task(chat_ns.process_messages())
# Send some messages
await chat_ns.send_message('Hello async world!')
await asyncio.sleep(1)
await chat_ns.send_message('How is everyone?')
# Wait for events
await sio.wait()
# Run the async client
asyncio.run(main())Namespace Registration
Namespaces must be registered with their respective servers or clients:
# Server namespace registration
sio = socketio.Server()
sio.register_namespace(ChatNamespace('/chat'))
sio.register_namespace(GameNamespace('/game'))
# Client namespace registration
client = socketio.Client()
client.register_namespace(ChatClientNamespace('/chat'))
client.register_namespace(GameClientNamespace('/game'))
# Async server namespace registration
async_sio = socketio.AsyncServer()
async_sio.register_namespace(AsyncChatNamespace('/chat'))
# Async client namespace registration
async_client = socketio.AsyncClient()
async_client.register_namespace(AsyncChatClientNamespace('/chat'))Install with Tessl CLI
npx tessl i tessl/pypi-python-socketio