Metadata-Version: 2.4
Name: zndraw-socketio
Version: 0.1.0
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: Atomie CHEN <atomic_cwh@163.com>, Fabian Zills <fabian.zills@web.de>
License-File: LICENSE
Keywords: pydantic,socketio,typed,wrapper,zndraw
Classifier: License :: OSI Approved :: MIT License
Classifier: Operating System :: OS Independent
Classifier: Programming Language :: Python :: 3
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

Typed wrapper for [python-socketio](https://github.com/miguelgrinberg/python-socketio) with Pydantic validation and dependency injection.


## Installation

```sh
pip install zndraw-socketio
```

Optional extras for client transports:

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


## Usage

### Wrap any socketio instance

```python
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

```python
# 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

```python
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

```python
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

```python
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

```python
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`:

```python
from zndraw_socketio import Depends
```

### FastAPI integration

```python
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

```python
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
