CtrlK
BlogDocsLog inGet started
Tessl Logo

tessl/pypi-python-engineio

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

—

Pending
Quality

Pending

Does it follow best practices?

Impact

Pending

No eval scenarios have been run

SecuritybySnyk

Pending

The risk profile of this skill

Overview
Eval results
Files

Python Engine.IO

A comprehensive Python implementation of the Engine.IO realtime communication protocol, providing both client and server capabilities for building real-time web applications. Engine.IO enables bidirectional event-based communication between clients and servers with automatic reconnection, transport fallbacks (WebSocket to HTTP long-polling), and seamless integration with multiple async frameworks.

Package Information

  • Package Name: python-engineio
  • Language: Python
  • Installation: pip install python-engineio
  • Python Requirements: >= 3.6
  • Core Dependencies: simple-websocket >= 0.10.0

Core Imports

import engineio

Common usage patterns:

# Server applications
from engineio import Server, AsyncServer
from engineio import WSGIApp, ASGIApp

# Client applications  
from engineio import Client, AsyncClient

# Exception handling
from engineio.exceptions import EngineIOError, ConnectionError

Basic Usage

Simple Server Example

import engineio

# Create synchronous 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'Received {data} from {sid}')
    eio.send(sid, 'response message')

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

# Use with WSGI application
app = engineio.WSGIApp(eio)

Simple Client Example

import engineio

# Create synchronous client
eio = engineio.Client()

@eio.on('connect')
def on_connect():
    print('Connected to server')
    eio.send('Hello Server!')

@eio.on('message')
def on_message(data):
    print(f'Received from server: {data}')

@eio.on('disconnect')
def on_disconnect():
    print('Disconnected from server')

# Connect to server
eio.connect('http://localhost:5000')
eio.wait()

Async Server Example

import engineio

# Create asynchronous server
eio = engineio.AsyncServer()

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

@eio.on('message')
async def on_message(sid, data):
    print(f'Received {data} from {sid}')
    await eio.send(sid, 'async response')

# Use with ASGI application
app = engineio.ASGIApp(eio)

Architecture

Engine.IO provides a layered architecture for real-time communication:

  • Transport Layer: Handles WebSocket and HTTP long-polling protocols with automatic upgrades
  • Session Management: Manages client connections, sessions, and automatic reconnection
  • Event System: Event-driven programming model for handling connections, messages, and disconnections
  • Middleware Integration: WSGI/ASGI middleware for seamless web framework integration
  • Async Support: Native support for asyncio, eventlet, gevent, and threading models

The library serves as the foundation transport layer for Socket.IO and other real-time communication systems, providing reliable bidirectional messaging with automatic connection management and protocol negotiation.

Capabilities

Synchronous Server

Full-featured Engine.IO server for synchronous Python applications with support for multiple async models including threading, eventlet, and gevent.

class Server:
    def __init__(self, async_mode=None, ping_interval=25, ping_timeout=20, **kwargs): ...
    def send(self, sid, data): ...
    def disconnect(self, sid=None): ...
    def on(self, event, handler=None): ...
    def handle_request(self, environ, start_response): ...

Server

Asynchronous Server

Engine.IO server optimized for asyncio-based applications with full async/await support and integration with modern async web frameworks.

class AsyncServer:
    def __init__(self, async_mode=None, ping_interval=25, ping_timeout=20, **kwargs): ...
    async def send(self, sid, data): ...
    async def disconnect(self, sid=None): ...
    def on(self, event, handler=None): ...
    async def handle_request(self, *args, **kwargs): ...

Async Server

Synchronous Client

Engine.IO client for connecting to servers from synchronous Python applications with automatic reconnection and transport fallback support.

class Client:
    def __init__(self, logger=False, request_timeout=5, **kwargs): ...
    def connect(self, url, headers=None, transports=None, engineio_path='engine.io'): ...
    def send(self, data): ...
    def disconnect(self, abort=False, reason=None): ...
    def on(self, event, handler=None): ...

Client

Asynchronous Client

Asyncio-based Engine.IO client for connecting to servers from async Python applications with full async/await support.

class AsyncClient:
    def __init__(self, logger=False, request_timeout=5, **kwargs): ...
    async def connect(self, url, headers=None, transports=None, engineio_path='engine.io'): ...
    async def send(self, data): ...
    async def disconnect(self, abort=False, reason=None): ...
    def on(self, event, handler=None): ...

Async Client

WSGI Middleware

WSGI application middleware for integrating Engine.IO servers with Flask, Django, and other WSGI-compatible web frameworks.

class WSGIApp:
    def __init__(self, engineio_app, wsgi_app=None, static_files=None, engineio_path='engine.io'): ...
    def __call__(self, environ, start_response): ...

WSGI Middleware

ASGI Middleware

ASGI application middleware for integrating Engine.IO servers with FastAPI, Starlette, and other ASGI-compatible web frameworks.

class ASGIApp:
    def __init__(self, engineio_server, other_asgi_app=None, static_files=None, **kwargs): ...
    async def __call__(self, scope, receive, send): ...

ASGI Middleware

Tornado Integration

Tornado WebSocket handler for integrating Engine.IO with Tornado web applications. Returns a dynamically created handler class that extends tornado.websocket.WebSocketHandler.

def get_tornado_handler(engineio_server):
    """
    Return a Tornado WebSocket handler class for Engine.IO integration.
    
    Args:
        engineio_server (AsyncServer): The Engine.IO async server instance
        
    Returns:
        class: Tornado WebSocket handler class that extends WebSocketHandler
        
    Note:
        Function may return None if Tornado is not installed.
        The returned handler class supports both HTTP and WebSocket requests.
    """

Exception Handling

Comprehensive exception system for handling Engine.IO-specific errors and connection issues.

class EngineIOError(Exception): ...
class ContentTooLongError(EngineIOError): ...
class UnknownPacketError(EngineIOError): ...
class QueueEmpty(EngineIOError): ...
class SocketIsClosedError(EngineIOError): ...
class ConnectionError(EngineIOError): ...

Exception Handling

Types

# Session ID type
SessionId = str

# Event handler signature
EventHandler = Callable[[str, Any], None]
AsyncEventHandler = Callable[[str, Any], Awaitable[None]]

# WSGI environment
WSGIEnviron = Dict[str, Any]

# Transport types
Transport = Literal['polling', 'websocket']
Workspace
tessl
Visibility
Public
Created
Last updated
Describes
pypipkg:pypi/python-engineio@4.12.x
Publish Source
CLI
Badge
tessl/pypi-python-engineio badge