tessl/pypi-python-socketio
Socket.IO server and client for Python providing real-time bidirectional communication
—
Pending
clients.mddocs/
Client Classes
Socket.IO client implementations for connecting to servers and exchanging real-time messages. The library provides both simple clients for basic use cases and full-featured clients with advanced capabilities like automatic reconnection, namespace support, and event-driven architecture.
Capabilities
SimpleClient
A simple synchronous Socket.IO client that provides blocking methods for basic real-time communication. Ideal for simple scripts and straightforward client scenarios.
class SimpleClient:
"""
Simple synchronous Socket.IO client with blocking methods.
Attributes:
sid (str): Session ID assigned by server
transport (str): Current transport method ('websocket' or 'polling')
"""
def connect(self, url, headers=None, auth=None, transports=None, wait_timeout=1):
"""
Connect to a Socket.IO server.
Args:
url (str): Server URL
headers (dict, optional): HTTP headers to send
auth (dict, optional): Authentication credentials
transports (list, optional): Allowed transport methods ['websocket', 'polling']
wait_timeout (int): Connection timeout in seconds
Raises:
ConnectionError: Failed to connect to server
"""
def emit(self, event, data=None):
"""
Emit an event to the server.
Args:
event (str): Event name
data: Event data to send
Raises:
DisconnectedError: Client not connected to server
"""
def call(self, event, data=None, timeout=60):
"""
Emit an event and wait for a response.
Args:
event (str): Event name
data: Event data to send
timeout (int): Response timeout in seconds
Returns:
Response data from server
Raises:
TimeoutError: No response received within timeout
DisconnectedError: Client not connected to server
"""
def receive(self, timeout=None):
"""
Wait for an event from the server.
Args:
timeout (int, optional): Receive timeout in seconds
Returns:
tuple: (event_name, event_data)
Raises:
TimeoutError: No event received within timeout
DisconnectedError: Client not connected to server
"""
def disconnect(self):
"""
Disconnect from the server.
"""Usage Example
import socketio
# Create and use simple client
sio = socketio.SimpleClient()
try:
# Connect to server
sio.connect('http://localhost:5000')
# Send a message
sio.emit('hello', {'name': 'Alice'})
# Wait for response
event, data = sio.receive(timeout=10)
print(f'Received {event}: {data}')
# Call with response
response = sio.call('get_status', timeout=5)
print(f'Status: {response}')
finally:
sio.disconnect()Client
A full-featured synchronous Socket.IO client with event-driven architecture, automatic reconnection, namespace support, and comprehensive configuration options.
class Client:
"""
Full-featured synchronous Socket.IO client.
Inherits from: BaseClient
Attributes:
sid (str): Session ID assigned by server
transport (str): Current transport method
connected (bool): Connection status
"""
def __init__(self, reconnection=True, reconnection_attempts=0, reconnection_delay=1,
reconnection_delay_max=5, randomization_factor=0.5, logger=False,
serializer='default', json=None, handle_sigint=True, **kwargs):
"""
Initialize the client.
Args:
reconnection (bool): Enable automatic reconnection
reconnection_attempts (int): Maximum reconnection attempts (0 = unlimited)
reconnection_delay (int): Initial reconnection delay in seconds
reconnection_delay_max (int): Maximum reconnection delay in seconds
randomization_factor (float): Randomization factor for reconnection delay
logger (bool or Logger): Enable logging or provide custom logger
serializer (str): Message serializer ('default', 'pickle', 'msgpack', 'cbor')
json (module): Custom JSON module
handle_sigint (bool): Handle SIGINT signal for graceful shutdown
**kwargs: Additional Engine.IO client parameters
"""
def connect(self, url, headers=None, auth=None, transports=None, namespaces=None,
socketio_path='socket.io', wait=True, wait_timeout=1):
"""
Connect to a Socket.IO server.
Args:
url (str): Server URL
headers (dict, optional): HTTP headers to send
auth (dict, optional): Authentication credentials
transports (list, optional): Allowed transport methods
namespaces (list, optional): Namespaces to connect to
socketio_path (str): Socket.IO endpoint path
wait (bool): Wait for connection to complete
wait_timeout (int): Connection timeout in seconds
Raises:
ConnectionError: Failed to connect to server
"""
def wait(self):
"""
Wait until the connection ends.
"""
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, optional): Target namespace
callback (callable, optional): Callback function for response
Raises:
DisconnectedError: Client not connected to server
"""
def send(self, data, namespace=None, callback=None):
"""
Send a message event to the server.
Args:
data: Message data to send
namespace (str, optional): Target namespace
callback (callable, optional): Callback function for response
Raises:
DisconnectedError: Client not connected to server
"""
def call(self, event, data=None, namespace=None, timeout=60):
"""
Emit an event and wait for a response.
Args:
event (str): Event name
data: Event data to send
namespace (str, optional): Target namespace
timeout (int): Response timeout in seconds
Returns:
Response data from server
Raises:
TimeoutError: No response received within timeout
DisconnectedError: Client not connected to server
"""
def disconnect(self):
"""
Disconnect from the server.
"""
def shutdown(self):
"""
Stop the client and all reconnection attempts.
"""
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, optional): Event handler function
namespace (str, optional): 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, optional): Event handler function
namespace (str, optional): Target namespace
Returns:
Decorator function
"""Usage Example
import socketio
# Create client with configuration
sio = socketio.Client(
reconnection=True,
reconnection_attempts=5,
reconnection_delay=2,
logger=True
)
# Register event handlers
@sio.event
def connect():
print('Connected to server')
sio.emit('join_room', {'room': 'general'})
@sio.event
def disconnect():
print('Disconnected from server')
@sio.event
def message(data):
print(f'Received message: {data}')
@sio.on('notification')
def handle_notification(data):
print(f'Notification: {data}')
# Connect and wait
sio.connect('http://localhost:5000')
# Emit with callback
def response_handler(data):
print(f'Server responded: {data}')
sio.emit('chat_message', {'text': 'Hello world!'}, callback=response_handler)
# Wait for events
sio.wait()AsyncSimpleClient
Asynchronous version of SimpleClient using coroutines for non-blocking operations. Suitable for asyncio-based applications requiring simple Socket.IO communication.
class AsyncSimpleClient:
"""
Simple asynchronous Socket.IO client with coroutine methods.
Attributes:
sid (str): Session ID assigned by server
transport (str): Current transport method
"""
async def connect(self, url, headers=None, auth=None, transports=None, wait_timeout=1):
"""
Connect to a Socket.IO server.
Args:
url (str): Server URL
headers (dict, optional): HTTP headers to send
auth (dict, optional): Authentication credentials
transports (list, optional): Allowed transport methods
wait_timeout (int): Connection timeout in seconds
Raises:
ConnectionError: Failed to connect to server
"""
async def emit(self, event, data=None):
"""
Emit an event to the server.
Args:
event (str): Event name
data: Event data to send
Raises:
DisconnectedError: Client not connected to server
"""
async def call(self, event, data=None, timeout=60):
"""
Emit an event and wait for a response.
Args:
event (str): Event name
data: Event data to send
timeout (int): Response timeout in seconds
Returns:
Response data from server
Raises:
TimeoutError: No response received within timeout
DisconnectedError: Client not connected to server
"""
async def receive(self, timeout=None):
"""
Wait for an event from the server.
Args:
timeout (int, optional): Receive timeout in seconds
Returns:
tuple: (event_name, event_data)
Raises:
TimeoutError: No event received within timeout
DisconnectedError: Client not connected to server
"""
async def disconnect(self):
"""
Disconnect from the server.
"""Usage Example
import asyncio
import socketio
async def main():
sio = socketio.AsyncSimpleClient()
try:
# Connect to server
await sio.connect('http://localhost:5000')
# Send message
await sio.emit('hello', {'name': 'Bob'})
# Wait for response
event, data = await sio.receive(timeout=10)
print(f'Received {event}: {data}')
# Call with response
response = await sio.call('get_info', {'query': 'status'})
print(f'Info: {response}')
finally:
await sio.disconnect()
# Run the async client
asyncio.run(main())AsyncClient
Full-featured asynchronous Socket.IO client with all the capabilities of the synchronous Client but using coroutines for non-blocking operation in asyncio applications.
class AsyncClient:
"""
Full-featured asynchronous Socket.IO client for asyncio.
Inherits from: BaseClient
Attributes:
sid (str): Session ID assigned by server
transport (str): Current transport method
connected (bool): Connection status
"""
def __init__(self, reconnection=True, reconnection_attempts=0, reconnection_delay=1,
reconnection_delay_max=5, randomization_factor=0.5, logger=False,
serializer='default', json=None, **kwargs):
"""
Initialize the async client.
Args:
reconnection (bool): Enable automatic reconnection
reconnection_attempts (int): Maximum reconnection attempts (0 = unlimited)
reconnection_delay (int): Initial reconnection delay in seconds
reconnection_delay_max (int): Maximum reconnection delay in seconds
randomization_factor (float): Randomization factor for reconnection delay
logger (bool or Logger): Enable logging or provide custom logger
serializer (str): Message serializer ('default', 'pickle', 'msgpack', 'cbor')
json (module): Custom JSON module
**kwargs: Additional Engine.IO client parameters
"""
async def connect(self, url, headers=None, auth=None, transports=None, namespaces=None,
socketio_path='socket.io', wait=True, wait_timeout=1):
"""
Connect to a Socket.IO server.
Args:
url (str): Server URL
headers (dict, optional): HTTP headers to send
auth (dict, optional): Authentication credentials
transports (list, optional): Allowed transport methods
namespaces (list, optional): Namespaces to connect to
socketio_path (str): Socket.IO endpoint path
wait (bool): Wait for connection to complete
wait_timeout (int): Connection timeout in seconds
Raises:
ConnectionError: Failed to connect to server
"""
async def wait(self):
"""
Wait until the connection ends.
"""
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, optional): Target namespace
callback (callable, optional): Async callback function for response
Raises:
DisconnectedError: Client not connected to server
"""
async def send(self, data, namespace=None, callback=None):
"""
Send a message event to the server.
Args:
data: Message data to send
namespace (str, optional): Target namespace
callback (callable, optional): Async callback function for response
Raises:
DisconnectedError: Client not connected to server
"""
async def call(self, event, data=None, namespace=None, timeout=60):
"""
Emit an event and wait for a response.
Args:
event (str): Event name
data: Event data to send
namespace (str, optional): Target namespace
timeout (int): Response timeout in seconds
Returns:
Response data from server
Raises:
TimeoutError: No response received within timeout
DisconnectedError: Client not connected to server
"""
async def disconnect(self):
"""
Disconnect from the server.
"""
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, optional): Async event handler function
namespace (str, optional): 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, optional): Async event handler function
namespace (str, optional): Target namespace
Returns:
Decorator function
"""Usage Example
import asyncio
import socketio
# Create async client
sio = socketio.AsyncClient(
reconnection=True,
reconnection_attempts=3,
logger=True
)
# Register async event handlers
@sio.event
async def connect():
print('Connected to server')
await sio.emit('join_room', {'room': 'async-users'})
@sio.event
async def disconnect():
print('Disconnected from server')
@sio.event
async def message(data):
print(f'Received message: {data}')
# Process message asynchronously
await process_message(data)
@sio.on('notification')
async def handle_notification(data):
print(f'Notification: {data}')
async def process_message(data):
# Simulate async processing
await asyncio.sleep(0.1)
print(f'Processed: {data}')
async def main():
# Connect to server
await sio.connect('http://localhost:5000')
# Emit with async callback
async def response_handler(data):
print(f'Server responded: {data}')
await sio.emit('chat_message', {'text': 'Hello async world!'}, callback=response_handler)
# Wait for events
await sio.wait()
# Run the async client
asyncio.run(main())Context Manager Support
Both SimpleClient and AsyncSimpleClient support context manager syntax for automatic connection management:
# Synchronous context manager
with socketio.SimpleClient() as sio:
sio.connect('http://localhost:5000')
sio.emit('hello', {'name': 'World'})
event, data = sio.receive()
# Asynchronous context manager
async with socketio.AsyncSimpleClient() as sio:
await sio.connect('http://localhost:5000')
await sio.emit('hello', {'name': 'World'})
event, data = await sio.receive()Install with Tessl CLI
npx tessl i tessl/pypi-python-socketio