CtrlK
BlogDocsLog inGet started
Tessl Logo

tessl/pypi-python-socketio

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

—

Pending
Overview
Eval results
Files

integration.mddocs/

Web Framework Integration

Middleware classes and utilities for integrating Socket.IO servers with popular Python web frameworks. These integration layers enable Socket.IO to work seamlessly with WSGI and ASGI applications while providing static file serving and flexible routing options.

Capabilities

WSGIApp

WSGI middleware that integrates a Socket.IO server with WSGI-compatible web frameworks like Flask, Django, and others.

class WSGIApp:
    """
    WSGI middleware for Socket.IO applications.
    
    Inherits from: engineio.WSGIApp
    
    Attributes:
        socketio_app: Socket.IO server instance
        wsgi_app: Wrapped WSGI application
        socketio_path (str): Socket.IO endpoint path
    """
    
    def __init__(self, socketio_app, wsgi_app=None, static_files=None, socketio_path='socket.io'):
        """
        Initialize the WSGI middleware.
        
        Args:
            socketio_app: Socket.IO server instance
            wsgi_app: WSGI application to wrap (optional)
            static_files (dict): Static file mappings {'/path': 'directory'}
            socketio_path (str): Socket.IO endpoint path (default: 'socket.io')
        """
    
    def __call__(self, environ, start_response):
        """
        WSGI application callable.
        
        Args:
            environ (dict): WSGI environment
            start_response (callable): WSGI start_response function
            
        Returns:
            WSGI response iterable
        """

Usage Examples

Standalone Socket.IO Application
import socketio

# Create Socket.IO server
sio = socketio.Server(cors_allowed_origins="*")

@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):
    sio.emit('response', {'echo': data})

# Create WSGI app with static files
app = socketio.WSGIApp(sio, static_files={
    '/': 'static/',
    '/css': 'static/css/',
    '/js': 'static/js/'
})

if __name__ == '__main__':
    import eventlet
    import eventlet.wsgi
    eventlet.wsgi.server(eventlet.listen(('', 5000)), app)
Flask Integration
import socketio
from flask import Flask

# Create Flask app
flask_app = Flask(__name__)

# Create Socket.IO server
sio = socketio.Server(cors_allowed_origins="*")

@flask_app.route('/')
def index():
    return '<h1>Flask + Socket.IO</h1>'

@flask_app.route('/api/data')
def get_data():
    return {'message': 'Hello from Flask!'}

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

@sio.event
def flask_message(sid, data):
    # Can access Flask app context if needed
    with flask_app.app_context():
        # Flask operations here
        pass
    sio.emit('flask_response', {'processed': data})

# Wrap Flask app with Socket.IO
app = socketio.WSGIApp(sio, flask_app)

if __name__ == '__main__':
    import eventlet
    import eventlet.wsgi
    eventlet.wsgi.server(eventlet.listen(('', 5000)), app)
Django Integration
import socketio
import os
from django.core.wsgi import get_wsgi_application

# Django setup
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'myproject.settings')
django_app = get_wsgi_application()

# Create Socket.IO server
sio = socketio.Server(cors_allowed_origins="*")

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

@sio.event
def django_message(sid, data):
    # Can import and use Django models
    from myapp.models import Message
    message = Message.objects.create(content=data['text'])
    sio.emit('message_saved', {'id': message.id})

# Combine Django + Socket.IO
app = socketio.WSGIApp(sio, django_app)
Custom Routing
import socketio

sio = socketio.Server()

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

# Custom Socket.IO path
app = socketio.WSGIApp(sio, socketio_path='custom-socketio')

# Client would connect to: http://localhost:5000/custom-socketio/

ASGIApp

ASGI middleware that integrates an AsyncServer with ASGI-compatible frameworks like FastAPI, Starlette, and Django Channels.

class ASGIApp:
    """
    ASGI middleware for async Socket.IO applications.
    
    Inherits from: engineio.ASGIApp
    
    Attributes:
        socketio_server: AsyncServer instance
        other_asgi_app: Wrapped ASGI application
        socketio_path (str): Socket.IO endpoint path
    """
    
    def __init__(self, socketio_server, other_asgi_app=None, static_files=None, 
                 socketio_path='socket.io', on_startup=None, on_shutdown=None):
        """
        Initialize the ASGI middleware.
        
        Args:
            socketio_server: AsyncServer instance
            other_asgi_app: ASGI application to wrap (optional)
            static_files (dict): Static file mappings {'/path': 'directory'}
            socketio_path (str): Socket.IO endpoint path (default: 'socket.io')
            on_startup (list): List of startup event handlers
            on_shutdown (list): List of shutdown event handlers
        """
    
    async def __call__(self, scope, receive, send):
        """
        ASGI application callable.
        
        Args:
            scope (dict): ASGI connection scope
            receive (callable): ASGI receive function
            send (callable): ASGI send function
        """

Usage Examples

Standalone Async Socket.IO Application
import socketio

# Create async Socket.IO server
sio = socketio.AsyncServer(cors_allowed_origins="*")

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

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

@sio.event
async def async_message(sid, data):
    # Simulate async processing
    await asyncio.sleep(0.1)
    await sio.emit('async_response', {'processed': data})

# Create ASGI app
app = socketio.ASGIApp(sio, static_files={
    '/': 'static/',
})

# Run with: uvicorn main:app --host 0.0.0.0 --port 5000
FastAPI Integration
import socketio
from fastapi import FastAPI

# Create FastAPI app
fastapi_app = FastAPI()

# Create async Socket.IO server
sio = socketio.AsyncServer(cors_allowed_origins="*")

@fastapi_app.get("/")
async def read_root():
    return {"message": "FastAPI + Socket.IO"}

@fastapi_app.get("/api/users/{user_id}")
async def get_user(user_id: int):
    return {"user_id": user_id, "name": f"User {user_id}"}

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

@sio.event
async def fastapi_message(sid, data):
    # Can use FastAPI dependencies and features
    await sio.emit('fastapi_response', {'echo': data})

# Combine FastAPI + Socket.IO
app = socketio.ASGIApp(sio, fastapi_app)

# Run with: uvicorn main:app --host 0.0.0.0 --port 5000
Starlette Integration
import socketio
from starlette.applications import Starlette
from starlette.responses import JSONResponse
from starlette.routing import Route

# Create async Socket.IO server
sio = socketio.AsyncServer(cors_allowed_origins="*")

async def homepage(request):
    return JSONResponse({'message': 'Starlette + Socket.IO'})

async def api_endpoint(request):
    return JSONResponse({'data': 'API response'})

# Create Starlette app
starlette_app = Starlette(routes=[
    Route('/', homepage),
    Route('/api/data', api_endpoint),
])

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

@sio.event
async def starlette_message(sid, data):
    await sio.emit('starlette_response', {'received': data})

# Combine Starlette + Socket.IO
app = socketio.ASGIApp(sio, starlette_app)
Startup and Shutdown Handlers
import socketio
from fastapi import FastAPI

sio = socketio.AsyncServer()
fastapi_app = FastAPI()

# Startup and shutdown handlers
async def startup_handler():
    print("Application starting up...")
    # Initialize resources, database connections, etc.

async def shutdown_handler():
    print("Application shutting down...")
    # Clean up resources

app = socketio.ASGIApp(
    sio, 
    fastapi_app,
    on_startup=[startup_handler],
    on_shutdown=[shutdown_handler]
)

get_tornado_handler

Utility function to create a Tornado request handler for Socket.IO server integration.

def get_tornado_handler(socketio_server):
    """
    Get Tornado request handler for Socket.IO server.
    
    Args:
        socketio_server: Socket.IO server instance (Server or AsyncServer)
        
    Returns:
        Tornado handler class for Socket.IO integration
    """

Usage Example

import socketio
import tornado.web
import tornado.ioloop

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

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

@sio.event
def tornado_message(sid, data):
    sio.emit('tornado_response', {'echo': data})

# Create Tornado application
class MainHandler(tornado.web.RequestHandler):
    def get(self):
        self.write("Tornado + Socket.IO")

# Get Socket.IO handler for Tornado
SocketIOHandler = socketio.get_tornado_handler(sio)

app = tornado.web.Application([
    (r"/", MainHandler),
    (r"/socket.io/.*", SocketIOHandler),
], debug=True)

if __name__ == "__main__":
    app.listen(5000)
    print("Server running on http://localhost:5000")
    tornado.ioloop.IOLoop.current().start()

Middleware (Deprecated)

class Middleware:
    """
    Deprecated alias for WSGIApp.
    
    Note:
        This class is deprecated. Use WSGIApp instead.
    """

The Middleware class is a deprecated alias for WSGIApp. Use WSGIApp directly for new applications.

Static File Serving

Both WSGIApp and ASGIApp support static file serving for client-side assets:

# Static file mappings
static_files = {
    '/': 'public/',           # Serve public/ directory at root
    '/css': 'assets/css/',    # Serve CSS from assets/css/
    '/js': 'assets/js/',      # Serve JavaScript from assets/js/
    '/images': 'assets/img/'  # Serve images from assets/img/
}

# WSGI with static files
wsgi_app = socketio.WSGIApp(sio, flask_app, static_files=static_files)

# ASGI with static files
asgi_app = socketio.ASGIApp(sio, fastapi_app, static_files=static_files)

CORS Configuration

Configure CORS (Cross-Origin Resource Sharing) for web browser clients:

# Allow all origins (development only)
sio = socketio.Server(cors_allowed_origins="*")

# Allow specific origins
sio = socketio.Server(cors_allowed_origins=[
    "http://localhost:3000",
    "https://myapp.com"
])

# Async server CORS
async_sio = socketio.AsyncServer(cors_allowed_origins=[
    "http://localhost:8080",
    "https://app.example.com"
])

Custom Socket.IO Endpoints

Change the default Socket.IO endpoint path:

# Custom endpoint path
app = socketio.WSGIApp(sio, socketio_path='realtime')
# Clients connect to: /realtime/

app = socketio.ASGIApp(sio, socketio_path='ws')
# Clients connect to: /ws/

Production Deployment

WSGI Deployment

# With Gunicorn + eventlet
# gunicorn --worker-class eventlet -w 1 main:app

# With uWSGI + gevent
# uwsgi --http :5000 --gevent 1000 --module main:app

ASGI Deployment

# With Uvicorn
# uvicorn main:app --host 0.0.0.0 --port 5000

# With Hypercorn
# hypercorn main:app --bind 0.0.0.0:5000

# With Daphne (Django Channels)
# daphne -b 0.0.0.0 -p 5000 main:app

Install with Tessl CLI

npx tessl i tessl/pypi-python-socketio

docs

clients.md

exceptions.md

index.md

integration.md

managers.md

namespaces.md

servers.md

tile.json