fastsio 0.3.3

light and typization Socket.IO server and client for Python

fastsio (Fast Socket.IO for Python)

fastsio is a fork of python-socketio with modern Python development improvements inspired by FastAPI.

Key Features

• Strong typing and FastAPI-like DX with dependency injection
• Automatic validation of data through Pydantic models
• Lightweight routers for code organization via RouterSIO
• Flexible middleware system for authentication, logging, rate limiting, and more
• Compatibility with existing ASGI/WSGI stacks
• Full support for the Socket.IO protocol

Installation

pip install fastsio

Key Differences from python-socketio

Feature	python-socketio	fastsio
Type Safety	Minimal	Strong typing with annotations
Dependency Injection	❌	✅ FastAPI-style
Pydantic Validation	❌	✅ Automatic
Routers	❌	✅ RouterSIO for code organization
Middlewares	❌	✅ Flexible middleware system
Parameter Annotations	❌	✅ SocketID, Environ, Auth, etc.
Compatibility	✅	✅ Full backward compatibility

Quick Start

Simple Server

import fastsio
from fastsio import ASGIApp, SocketID, Environ, Auth, Data

# Create server
sio = fastsio.AsyncServer(
    async_mode="asgi",
    cors_allowed_origins="*"
)

@sio.event
async def connect(
    sid: SocketID, 
    environ: Environ, 
    auth: Auth,
):
    print(f"Client {sid} connected")
    return True

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

@sio.on("message")
async def handle_message(sid: SocketID, data: Data):
    await sio.emit("response", f"Received: {data}", to=sid)

# ASGI application
app = ASGIApp(sio)

With Type Safety and Dependency Injection

from pydantic import BaseModel
from fastsio import (
    AsyncServer, RouterSIO, SocketID, 
    Environ, Auth, ASGIApp
)

router = RouterSIO()

class Message(BaseModel):
    text: str
    room: str

class JoinRoom(BaseModel):
    room: str

@router.on("connect")
async def on_connect(
    sid: SocketID, 
    environ: Environ, 
    auth: Auth, 
    server: AsyncServer
):
    """Connection with automatic dependency injection"""
    print(f"Connection: {sid}, auth: {auth}")
    return True

@router.on("join_room")
async def on_join_room(
    sid: SocketID, 
    server: AsyncServer, 
    data: JoinRoom  # Automatic Pydantic validation
):
    """Join room with data validation"""
    await server.enter_room(sid, data.room)
    await server.emit("joined", {"room": data.room}, to=sid)

@router.on("send_message")
async def on_send_message(
    sid: SocketID, 
    server: AsyncServer, 
    data: Message
):
    """Send message with validation"""
    await server.emit(
        "new_message", 
        data.model_dump(), 
        room=data.room
    )

# Create server and attach router
sio = AsyncServer(async_mode="asgi", cors_allowed_origins="*")
sio.add_router(router)

app = ASGIApp(sio)

FastAPI Integration

import uvicorn
from fastapi import FastAPI
from fastapi.responses import FileResponse
import fastsio
from fastsio import Environ, Auth, SocketID

app = FastAPI()
sio = fastsio.AsyncServer(async_mode="asgi")

@app.get("/")
async def index():
    return {"message": "FastAPI + fastsio"}

@sio.event
async def connect(sid: SocketID, environ: Environ, auth: Auth):
    await sio.emit("hello", {"message": "Welcome!"}, to=sid)

# Combine FastAPI and Socket.IO
combined_app = fastsio.ASGIApp(sio, app)

if __name__ == "__main__":
    uvicorn.run(combined_app, host="127.0.0.1", port=5000)

Advanced Features

Middlewares for Cross-Cutting Concerns

from typing import Any
from fastsio import AsyncServer, BaseMiddleware, auth_middleware, logging_middleware

# Custom authentication middleware
class AuthMiddleware(BaseMiddleware):
    def __init__(self):
        super().__init__(events=["join_room", "send_message"])
    
    async def before_event(self, event: str, sid: str, data: Any, environ: dict = None, **kwargs):
        if not environ or not environ.get("HTTP_AUTHORIZATION"):
            raise PermissionError("Authentication required")
        return data

# Create server with middlewares
sio = AsyncServer(async_mode="asgi")

# Add middlewares with different scopes
sio.add_middleware(logging_middleware(), global_middleware=True)  # All events
sio.add_middleware(AuthMiddleware())  # Specific events only
sio.add_middleware(
    logging_middleware(), 
    namespace="/admin", 
    events=["admin_action"]
)  # Namespace + event specific

@sio.event
async def join_room(sid: SocketID, data: dict):
    # This will go through AuthMiddleware
    return {"status": "joined"}

Routers for Code Organization

from fastsio import RouterSIO

# Chat router
chat_router = RouterSIO(namespace="/chat")

@chat_router.on("message")
async def chat_message(sid: SocketID, data: Message):
    # Chat logic
    pass

# Admin router
admin_router = RouterSIO(namespace="/admin")

@admin_router.on("user_ban")
async def ban_user(sid: SocketID, data: BanRequest):
    # Ban logic
    pass

# Attach routers
sio.add_router(chat_router)
sio.add_router(admin_router)

Available Types for Injection

from fastsio import (
    SocketID,    # Connection ID
    Environ,     # WSGI/ASGI environ
    Auth,        # Auth data (connect handler only)
    Reason,      # Disconnect reason (disconnect handler only)
    Data,        # Raw event data
    Event,       # Event name
    AsyncServer  # Server instance
)

@router.on("example")
async def example_handler(
    sid: SocketID,
    server: AsyncServer,
    environ: Environ,
    data: Data,
    event: Event
):
    print(f"Event {event} from {sid}: {data}")

Version Compatibility

Compatibility table with JavaScript Socket.IO:

JavaScript Socket.IO	Socket.IO protocol	Engine.IO protocol	fastsio version
0.9.x	1, 2	1, 2	Not supported
1.x and 2.x	3, 4	3	Any version
3.x and 4.x	5	4	Any version

Documentation

• Full Documentation
• PyPI
• Changelog

Contributing

fastsio is based on the excellent work by Miguel Grinberg on python-socketio. We've added modern capabilities while maintaining full compatibility with the original API.

License

MIT License - see LICENSE file

---

🌍 Other Languages: Русский