Metadata-Version: 2.4
Name: zndraw-socketio
Version: 0.1.6
Summary: Typed wrapper for python-socketio with Pydantic validation and dependency injection.
Project-URL: homepage, https://github.com/pythonFZ/zndraw-socketio
Project-URL: issues, https://github.com/pythonFZ/zndraw-socketio/issues
Author-email: Fabian Zills <fabian.zills@web.de>
License: MIT License
        
        Copyright (c) 2026 zndraw-socketio contributors
        
        Permission is hereby granted, free of charge, to any person obtaining a copy
        of this software and associated documentation files (the "Software"), to deal
        in the Software without restriction, including without limitation the rights
        to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
        copies of the Software, and to permit persons to whom the Software is
        furnished to do so, subject to the following conditions:
        
        The above copyright notice and this permission notice shall be included in all
        copies or substantial portions of the Software.
        
        THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
        IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
        FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
        AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
        LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
        OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
        SOFTWARE.
License-File: LICENSE
Keywords: pydantic,socketio,typed,wrapper,zndraw
Requires-Python: >=3.10
Requires-Dist: pydantic>=2.10.6
Requires-Dist: python-socketio>=5.12.1
Requires-Dist: typing-extensions>=4.13.0
Provides-Extra: asyncio-client
Requires-Dist: python-socketio[asyncio-client]>=5.12.1; extra == 'asyncio-client'
Provides-Extra: client
Requires-Dist: python-socketio[client]>=5.12.1; extra == 'client'
Description-Content-Type: text/markdown

# ZnDraw SocketIO
This package provides an opinionated typed interface to the python-socketio library using pydantic models.

```python
from zndraw_socketio import wrap
from pydantic import BaseModel
import socketio

sio = wrap(socketio.AsyncClient())  # or AsyncServer, Client, Server, etc.
```

## Emit Pattern
```python
class Ping(BaseModel):
    message: str

# kwargs are passed to socketio's emit method
# emits {"message": "Hello, World!"} to "ping"
await sio.emit(Ping(message="Hello, World!"), **kwargs)
# emits {"message": "Hello, World!"} to "my-ping"
await sio.emit("my-ping", Ping(message="Hello, World!"), **kwargs)
# standard sio behaviour
await sio.emit("event", {"payload": ...})
```

## Call / RPC Pattern
```python
class Pong(BaseModel):
    reply: str

# emits {"message": "Hello, World!"} to "ping" and receives Pong(reply=...) in return
response = await sio.call(Ping(message="Hello, World!"), response_model=Pong)
assert isinstance(response, Pong)
# emits {"message": "Hello, World!"} to "my-ping" and receives Pong(reply=...) in return
response = await sio.call("my-ping", Ping(message="Hello, World!"), response_model=Pong)
assert isinstance(response, Pong)
# standard sio behaviour
response = await sio.call("event", {"payload": ...})
# standard response obj, typically dict
```

## Handler Registration
Handlers are registered with `@sio.on()` or `@sio.event` and get automatic Pydantic validation from type hints.

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

@tsio.on(Ping)
async def handle_ping(sid: str, data: Ping) -> Pong:
    return Pong(reply=data.message)

# or use the function name as the event name
@tsio.event
async def ping(sid: str, data: Ping) -> Pong:
    return Pong(reply=data.message)
```

## Outbound Event Documentation
Handlers that emit events to other channels can declare them with `emits`:

```python
class SessionLeft(BaseModel):
    room_id: str
    user_id: str

@tsio.on("disconnect", emits=[SessionLeft])
async def handle_disconnect(sid: str) -> None:
    await tsio.emit(SessionLeft(room_id="room1", user_id="user1"), room="room1")
```

These appear as `action: "send"` operations in the generated AsyncAPI schema.

### REST Endpoint Emits
REST endpoints can also declare socket event emissions using the `Emits` annotation. These are auto-discovered when `tsio.app` is set and appear as `x-rest-triggers` in the AsyncAPI schema.

```python
from typing import Annotated
from fastapi import Depends, FastAPI
from zndraw_socketio import Emits, AsyncServerWrapper, wrap

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

@app.put("/{key}/selection")
async def update_selection(
    sio: Annotated[AsyncServerWrapper, Depends(tsio), Emits(SessionLeft)],
    key: str,
) -> dict:
    await sio.emit(SessionLeft(room_id=key, user_id="..."), room=key)
    return {"status": "ok"}
```

## AsyncAPI Schema Generation
Generate an AsyncAPI 3.0 specification from registered handlers:

```python
schema = tsio.asyncapi_schema(title="My API", version="1.0.0")
```

## Event Names
By default, the event name is the class name in snake_case. You can customize it by setting the `event_name` attribute.

```python
class CustomEvent(BaseModel):
    ...

get_event_name(CustomEvent) == "custom_event"
```

You can override it like this:

```python
from typing import ClassVar

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

## Dependency Injection (Depends)
Handlers support FastAPI-style `Depends()` for dependency injection.

```python
from typing import Annotated
from zndraw_socketio import wrap, Depends

async def get_redis() -> Redis:
    return Redis()

RedisDep = Annotated[Redis, Depends(get_redis)]

@tsio.on(Ping)
async def handle(sid: str, data: Ping, redis: RedisDep) -> Pong:
    await redis.set("last_ping", data.message)
    return Pong(reply=data.message)
```

### FastAPI App Integration
To use dependencies that need `request.app` (e.g. accessing `app.state`), set the `.app` property on the wrapper. The app can be set **after** handler registration — it is resolved at event time, not at registration time.

```python
from fastapi import FastAPI, Request

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

def get_db(request: Request) -> Database:
    return request.app.state.db

@tsio.on(Ping)
async def handle(sid: str, data: Ping, db: Annotated[Database, Depends(get_db)]) -> Pong:
    ...

# Set app later — e.g. in a lifespan, after handler registration
tsio.app = app
```

## Exception Handlers
Server wrappers support exception handlers similar to FastAPI.

```python
from zndraw_socketio import EventContext

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

## Union Return Types
You might want to return `Response | ErrorResponse` from an event handler.

> [!NOTE]
> If your responses share fields, it is recommended to add a discriminator field to avoid ambiguity.

```python
class ProblemDetail(BaseModel):
    """RFC 9457 Problem Details."""
    kind: Literal["error"] = "error"
    type: str = "about:blank"
    title: str
    status: int
    detail: str | None = None
    instance: str | None = None

class Response(BaseModel):
    kind: Literal["response"] = "response"
    data: str

class ServerRequest(BaseModel):
    query: str

response = await sio.call(
    ServerRequest(query="..."),
    response_model=Response | ProblemDetail,
)
```
