socketspec 0.1.1

FastAPI-style WebSocket framework with built-in docs and testing

SocketSpec

FastAPI-style WebSocket framework with built-in interactive docs and testing.

---

Install

pip install socketspec[fastapi]

---

Quickstart

from fastapi import FastAPI
from pydantic import BaseModel
from socketspec import SocketApp
from socketspec.adapters.fastapi import mount

socket = SocketApp(docs=True)

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

class MessageAck(BaseModel):
    status: str
    message_id: str

@socket.on(
    "send_message",
    description="Send a chat message to a room.",
    tags=["chat"],
    emits=[Emits("message_ack", model=MessageAck, description="Delivery confirmation")],
    broadcasts=[Broadcasts("new_message", room="chat:{room}", description="Delivered to room members")],
)
async def send_message(conn, payload: ChatMessage) -> None:
    await conn.emit("message_ack", {"status": "ok", "message_id": "abc123"})
    await socket.rooms.broadcast(
        "chat:" + payload.room,
        "new_message",
        {"from": conn.id, "text": payload.text},
    )

app = FastAPI()
mount(socket, app, path="/ws")

uvicorn main:app --reload

---

Docs UI

Open /socket-docs in your browser after starting the server.

• Click an event card to expand its schema
• Hit Try it out to send a live WebSocket message
• See the server response appear inline, without leaving the browser
• Open a second tab and connect as a different user to test broadcasts

---

Why SocketSpec

Feature	python-socketio	channels (Django)	SocketSpec
FastAPI-native	✗	✗	✅
Pydantic payload validation	✗	✗	✅
Built-in interactive docs	✗	✗	✅
TestClient for unit tests	✗	partial	✅
Room guards / permissions	manual	manual	✅
Dependency injection (Depends)	✗	✗	✅
Type-safe (mypy --strict)	✗	✗	✅

---

Core Concepts

Event handlers look exactly like FastAPI route handlers:

@socket.on("join_room", tags=["rooms"])
async def join_room(conn: Connection, payload: JoinPayload) -> None:
    await socket.rooms.join(conn, f"chat:{payload.room_id}")

Rooms with pattern-based guards:

@socket.room_guard("admin:{room}")
async def admin_only(conn: Connection, room: str) -> bool:
    return conn.identity.role == "admin"

Testing without a real server:

from socketspec.testing import TestClient

async def test_send_message():
    async with TestClient(socket) as client:
        conn = await client.connect()
        await conn.send("send_message", {"room": "general", "text": "hello"})
        response = await conn.receive()
        assert response["event"] == "message_ack"

---

Links

• Documentation
• GitHub
• PyPI
• Changelog
• Contributing

---

License

Apache 2.0 — see LICENSE. Created and maintained by Laiba Shahab.