tessl/pypi-python-engineio

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

—

Pending

Overview

Eval results

Files

WSGI Middleware

Name: tessl/pypi-python-engineio
Author: tessl

WSGI application middleware for integrating Engine.IO servers with Flask, Django, and other WSGI-compatible web frameworks. Provides seamless integration with existing web applications while handling Engine.IO traffic.

Capabilities

Middleware Initialization

Create WSGI middleware that wraps an Engine.IO server with optional static file serving and fallback application support.

class WSGIApp:
    def __init__(
        self,
        engineio_app,
        wsgi_app=None,
        static_files=None,
        engineio_path='engine.io'
    ):
        """
        Initialize WSGI middleware for Engine.IO.

        Args:
            engineio_app (Server): The Engine.IO server instance
            wsgi_app (callable, optional): WSGI app for non-Engine.IO traffic
            static_files (dict, optional): Static file mapping rules
            engineio_path (str): Engine.IO endpoint path, default 'engine.io'
        """

WSGI Application Interface

Standard WSGI application callable that routes requests between Engine.IO and fallback applications.

def __call__(self, environ, start_response):
    """
    WSGI application callable.

    Args:
        environ (dict): WSGI environment dictionary
        start_response (callable): WSGI start_response callable

    Returns:
        iterable: WSGI response iterable
    """

Error Handling

Built-in error handling for requests that don't match Engine.IO patterns.

def not_found(self, start_response):
    """
    Return a 404 Not Found response.

    Args:
        start_response (callable): WSGI start_response callable

    Returns:
        list: 404 response body
    """

Integration Examples

Flask Integration

import engineio
from flask import Flask

# Create Flask app
app = Flask(__name__)

# Create Engine.IO server
eio = engineio.Server()

@eio.on('connect')
def on_connect(sid, environ):
    print(f'Client {sid} connected')

@eio.on('message')
def on_message(sid, data):
    print(f'Message from {sid}: {data}')
    eio.send(sid, f'Echo: {data}')

@eio.on('disconnect')
def on_disconnect(sid):
    print(f'Client {sid} disconnected')

# Flask routes
@app.route('/')
def index():
    return '<h1>Hello World!</h1>'

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

# Wrap Flask app with Engine.IO middleware
app = engineio.WSGIApp(eio, app)

if __name__ == '__main__':
    import eventlet
    import eventlet.wsgi
    eventlet.wsgi.server(eventlet.listen(('', 5000)), app)

Django Integration

# django_project/wsgi.py
import os
import engineio
from django.core.wsgi import get_wsgi_application

os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'django_project.settings')

# Get Django WSGI application
django_app = get_wsgi_application()

# Create Engine.IO server
eio = engineio.Server()

@eio.on('connect')
def on_connect(sid, environ):
    print(f'Client {sid} connected')

@eio.on('message')
def on_message(sid, data):
    eio.send(sid, f'Django says: {data}')

# Wrap Django with Engine.IO middleware
application = engineio.WSGIApp(eio, django_app)

Standalone WSGI Application

import engineio

# Create Engine.IO server only (no fallback app)
eio = engineio.Server()

@eio.on('connect')
def on_connect(sid, environ):
    print(f'Client {sid} connected')

@eio.on('message')
def on_message(sid, data):
    eio.send(sid, data.upper())

# Create standalone WSGI app
app = engineio.WSGIApp(eio)

# Deploy with any WSGI server
if __name__ == '__main__':
    from wsgiref.simple_server import make_server
    
    server = make_server('localhost', 8000, app)
    print('Server running on http://localhost:8000')
    server.serve_forever()

Static File Serving

import engineio

eio = engineio.Server()

# Define static file mappings
static_files = {
    '/': 'index.html',
    '/static/(.*)': r'static/\1',
    '/images/(.*)': r'assets/images/\1'
}

# Create WSGI app with static file serving
app = engineio.WSGIApp(eio, static_files=static_files)

Static file mapping rules:

Keys are URL patterns (can include regex groups)
Values are file paths (can reference regex groups with \1, \2, etc.)
Files are served relative to the current working directory

Custom Endpoint Path

import engineio

eio = engineio.Server()

# Use custom Engine.IO endpoint
app = engineio.WSGIApp(eio, engineio_path='custom/socket.io')

# Clients should connect to: http://server/custom/socket.io/

Multiple Engine.IO Servers

import engineio

# Create multiple servers for different purposes
chat_server = engineio.Server()
notifications_server = engineio.Server()

@chat_server.on('message')
def on_chat_message(sid, data):
    chat_server.send(sid, f'Chat: {data}')

@notifications_server.on('message')
def on_notification(sid, data):
    notifications_server.send(sid, f'Notification: {data}')

# Create separate middleware for each server
chat_app = engineio.WSGIApp(chat_server, engineio_path='chat')
notifications_app = engineio.WSGIApp(notifications_server, engineio_path='notifications')

# Combine using URL routing (example with Werkzeug)
from werkzeug.wsgi import DispatcherMiddleware

application = DispatcherMiddleware(None, {
    '/chat': chat_app,
    '/notifications': notifications_app
})

Middleware Class (Deprecated)

class Middleware(WSGIApp):
    def __init__(self, engineio_app, wsgi_app=None, engineio_path='engine.io'):
        """
        Deprecated alias for WSGIApp.
        
        Args:
            engineio_app (Server): The Engine.IO server instance
            wsgi_app (callable, optional): WSGI app for non-Engine.IO traffic
            engineio_path (str): Engine.IO endpoint path, default 'engine.io'
            
        Note:
            This class has been renamed to WSGIApp and is now deprecated.
            Use WSGIApp instead for new applications. The static_files parameter
            is not supported in this deprecated class.
        """

WSGI Environment Access

Engine.IO event handlers receive the WSGI environment dictionary, providing access to request details:

@eio.on('connect')
def on_connect(sid, environ):
    # Access request information
    remote_addr = environ.get('REMOTE_ADDR')
    user_agent = environ.get('HTTP_USER_AGENT')
    request_method = environ.get('REQUEST_METHOD')
    query_string = environ.get('QUERY_STRING')
    
    print(f'Client {sid} connected from {remote_addr}')
    print(f'User-Agent: {user_agent}')
    
    # Parse query parameters
    from urllib.parse import parse_qs
    params = parse_qs(query_string)
    
    # Store client info in session
    eio.save_session(sid, {
        'remote_addr': remote_addr,
        'user_agent': user_agent,
        'params': params
    })

Deployment Considerations

Production Deployment

# For production, use a proper WSGI server
import engineio
from flask import Flask

app = Flask(__name__)
eio = engineio.Server(async_mode='eventlet')

# ... configure server and handlers ...

app = engineio.WSGIApp(eio, app)

# Deploy with Gunicorn + eventlet workers
# gunicorn --worker-class eventlet -w 1 --bind 0.0.0.0:5000 app:app

Threading Considerations

import engineio

# Configure for threading mode
eio = engineio.Server(async_mode='threading')

@eio.on('message')
def on_message(sid, data):
    # Handler will run in a separate thread
    import time
    time.sleep(1)  # Blocking operations are OK
    eio.send(sid, 'Response after delay')

app = engineio.WSGIApp(eio)

Eventlet Integration

import eventlet
eventlet.monkey_patch()

import engineio

# Eventlet mode for high concurrency
eio = engineio.Server(async_mode='eventlet')

@eio.on('message')
def on_message(sid, data):
    # Use eventlet.sleep for non-blocking delays
    eventlet.sleep(1)
    eio.send(sid, 'Eventlet response')

app = engineio.WSGIApp(eio)

# Run with eventlet WSGI server
eventlet.wsgi.server(eventlet.listen(('', 5000)), app)

Error Handling

The WSGI middleware handles various error conditions automatically:

Invalid Engine.IO requests: Returns appropriate HTTP error responses
Non-Engine.IO requests: Routes to fallback WSGI app or returns 404
Static file errors: Returns 404 for missing files
Server errors: Logs exceptions and returns 500 responses

Custom error handling can be implemented in the fallback WSGI application:

def custom_app(environ, start_response):
    """Custom WSGI app with error handling"""
    try:
        # Application logic here
        response_body = b'Custom response'
        status = '200 OK'
        headers = [('Content-Type', 'text/plain')]
    except Exception as e:
        # Custom error handling
        response_body = f'Error: {str(e)}'.encode()
        status = '500 Internal Server Error'
        headers = [('Content-Type', 'text/plain')]
    
    start_response(status, headers)
    return [response_body]

app = engineio.WSGIApp(eio, custom_app)

Install with Tessl CLI