zndraw-socketio 0.1.0

Typed wrapper for python-socketio with Pydantic validation and dependency injection.

zndraw-socketio

Typed wrapper for python-socketio with Pydantic validation and dependency injection.

Installation

pip install zndraw-socketio

Optional extras for client transports:

pip install zndraw-socketio[client]
pip install zndraw-socketio[asyncio-client]

Usage

Wrap any socketio instance

import socketio
from pydantic import BaseModel
from zndraw_socketio import wrap

class Ping(BaseModel):
    message: str

class Pong(BaseModel):
    reply: str

# Wrap any existing socketio instance
tsio = wrap(socketio.AsyncClient())

# Emit with automatic event name derivation (Ping -> "ping")
await tsio.emit(Ping(message="Hello, World!"))

# Emit with explicit event name
await tsio.emit("my-ping", Ping(message="Hello, World!"))

# Call with typed response
response = await tsio.call(Ping(message="Hello"), response_model=Pong)
# response is typed as Pong

Handler registration with validation

# Event name derived from model class (Ping -> "ping")
@tsio.on(Ping)
async def handle_ping(data: Ping) -> Pong:
    return Pong(reply=data.message)

# Or use function name as event name
@tsio.event
async def ping(data: Ping) -> Pong:
    return Pong(reply=data.message)

Union response types

from typing import Annotated, Literal
from pydantic import BaseModel, Discriminator

class Success(BaseModel):
    kind: Literal["success"] = "success"
    data: str

class Error(BaseModel):
    kind: Literal["error"] = "error"
    message: str

# Simple union
response = await tsio.call(request, response_model=Success | Error)

# Discriminated union
ResponseType = Annotated[Success | Error, Discriminator("kind")]
response = await tsio.call(request, response_model=ResponseType)

Custom event names

from typing import ClassVar

class CustomEvent(BaseModel):
    event_name: ClassVar[str] = "my_custom_event"
    data: str

# Uses "my_custom_event" instead of "custom_event"
await tsio.emit(CustomEvent(data="hello"))

Exception handlers

from zndraw_socketio import wrap, EventContext

tsio = wrap(socketio.AsyncServer(async_mode="asgi"))

@tsio.exception_handler(ValueError)
async def handle_value_error(ctx: EventContext, exc: ValueError):
    return {"error": "value_error", "message": str(exc)}

# Namespace-specific
@tsio.exception_handler(ValueError, namespace="/chat")
async def handle_chat_error(ctx: EventContext, exc: ValueError):
    return {"error": "chat_error", "message": str(exc)}

Dependency injection

from typing import Annotated
from fastapi import Depends
from zndraw_socketio import wrap

async def get_redis():
    return my_redis_pool

RedisDep = Annotated[AsyncRedis, Depends(get_redis)]

tsio = wrap(socketio.AsyncServer(async_mode="asgi"))

@tsio.on(RoomLeave)
async def room_leave(sid: str, data: RoomLeave, redis: RedisDep) -> RoomLeaveResponse:
    await redis.delete(f"presence:{data.room_id}:{sid}")
    return RoomLeaveResponse(status="ok")

When FastAPI is not installed, import Depends from zndraw_socketio:

from zndraw_socketio import Depends

FastAPI integration

import socketio
from typing import Annotated
from fastapi import FastAPI, Depends
from zndraw_socketio import wrap, AsyncServerWrapper

app = FastAPI()
tsio = wrap(socketio.AsyncServer(async_mode="asgi"))

SioServer = Annotated[AsyncServerWrapper, Depends(tsio)]

@app.post("/notify")
async def notify(server: SioServer):
    await server.emit("notification", {"msg": "hello"})
    return {"status": "sent"}

combined_app = socketio.ASGIApp(tsio, app)

SimpleClient support

tsio = wrap(socketio.SimpleClient())
tsio.connect("http://localhost:5000")

tsio.emit(Ping(message="Hello"))
response = tsio.call(Ping(message="Hello"), response_model=Pong)
event_name, data = tsio.receive(response_model=Pong)
tsio.disconnect()

Supported socketio types

wrap() auto-detects the socketio instance type:

socketio type	Wrapper class
AsyncClient	AsyncClientWrapper
AsyncServer	AsyncServerWrapper
Client	SyncClientWrapper
Server	SyncServerWrapper
SimpleClient	SimpleClientWrapper
AsyncSimpleClient	AsyncSimpleClientWrapper

License

MIT License