# Fastapi-socketio-handler
**FastAPI + Socket.IO integration made modular, extensible and simple.**
A clean event-based wrapper that helps you organize your Socket.IO server logic using decorators, handlers, and namespaces.
---
## 🔧 Features
- 📡 Socket.IO server for FastAPI apps
- 🧩 Handler registration via decorators
- 📁 Namespace-based routing
- 🔁 Redis pub/sub support (scaling)
- 💡 Typed, extensible, and testable architecture
- 🧪 Ready for pytest & async testing
---
## 📦 Installation
```shell
pip install fastapi-socketio-handler
```
## 🚀 Quick Start
### 1. Define a handler
```python
# app/chat_handler.py
from socketio_handler import BaseSocketHandler, register_handler
@register_handler(namespace="/chat")
class ChatSocketHandlers(BaseSocketHandler):
def register_events(self):
self.sio.on("typing", self.event_typing, namespace=self.namespace)
self.sio.on("stop_typing", self.event_stop_typing, namespace=self.namespace)
async def connect(self, sid: str, environ: dict, auth: dict = None):
if not auth or "token" not in auth:
return False # Reject connection
return True
async def event_typing(self, sid: str, data: dict, *args):
print(f'Typing event from {sid}: {data}', args)
async def event_stop_typing(self, sid: str, data: dict, *args):
print(f'StopTyping event from {sid}: {data}', args)
```
### 2. Use with lifespan (recommended)
```python
# core/db.py
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
engine = create_async_engine(url="database_uri", echo=True)
async_session = async_sessionmaker(bind=engine, expire_on_commit=False, autocommit=False, autoflush=False)
```
```python
# core/lifespan.py
from contextlib import asynccontextmanager
from typing import TYPE_CHECKING
from socketio_handler import SocketManager
from .db import async_session
import app.chat_handler # 👈 force-import handlers to trigger decorator registration
if TYPE_CHECKING:
from fastapi import FastAPI
@asynccontextmanager
async def lifespan(app: "FastAPI"):
"""
Lifespan context manager for FastAPI application.
Useful for initializing and cleaning up resources.
"""
async with (
SocketManager(
redis_url="redis://localhost:6379",
async_session=async_session,
) as socket_manager,
):
socket_manager.mount_to_app(app)
socket_manager.register_events()
app.state.socket_manager = socket_manager
yield
```
```python
# main.py
from fastapi import FastAPI
from core.lifespan import lifespan
app = FastAPI(lifespan=lifespan)
```
### 3. Connect from frontend
```javascript
const socket = io('http://localhost:8000/chat', {
auth: {
token: 'your-auth-token'
}
});
socket.emit("typing", { chatId: "..." });
```
## 🧪 Testing
This package includes pytest-based test utilities.
You can run the tests with:
* Install dependencies for testing
```shell
poetry install --with dev
```
* Run tests
```shell
pytest
```
Raw data
{
"_id": null,
"home_page": "https://github.com/bandirom/fastapi-socketio-handler",
"name": "fastapi-socketio-handler",
"maintainer": null,
"docs_url": null,
"requires_python": ">=3.9",
"maintainer_email": null,
"keywords": "socketio, asyncio, websockets, python-socketio, asyncio-socketio, websockets-handler, fastapi",
"author": "Nazarii",
"author_email": "bandirom@ukr.net",
"download_url": "https://files.pythonhosted.org/packages/4a/44/9885892ba8be84c59691cb87880b5eae6acf14920f688c8cf123158d705a/fastapi_socketio_handler-0.1.2.tar.gz",
"platform": null,
"description": "# Fastapi-socketio-handler\n\n**FastAPI + Socket.IO integration made modular, extensible and simple.**\n\nA clean event-based wrapper that helps you organize your Socket.IO server logic using decorators, handlers, and namespaces.\n\n---\n\n## \ud83d\udd27 Features\n\n- \ud83d\udce1 Socket.IO server for FastAPI apps\n- \ud83e\udde9 Handler registration via decorators\n- \ud83d\udcc1 Namespace-based routing\n- \ud83d\udd01 Redis pub/sub support (scaling)\n- \ud83d\udca1 Typed, extensible, and testable architecture\n- \ud83e\uddea Ready for pytest & async testing\n\n---\n\n## \ud83d\udce6 Installation\n\n```shell\npip install fastapi-socketio-handler\n```\n\n\n## \ud83d\ude80 Quick Start\n\n\n### 1. Define a handler\n\n```python\n# app/chat_handler.py\n\nfrom socketio_handler import BaseSocketHandler, register_handler\n\n\n@register_handler(namespace=\"/chat\")\nclass ChatSocketHandlers(BaseSocketHandler):\n\n def register_events(self):\n self.sio.on(\"typing\", self.event_typing, namespace=self.namespace)\n self.sio.on(\"stop_typing\", self.event_stop_typing, namespace=self.namespace)\n\n async def connect(self, sid: str, environ: dict, auth: dict = None):\n if not auth or \"token\" not in auth:\n return False # Reject connection\n return True\n\n async def event_typing(self, sid: str, data: dict, *args):\n print(f'Typing event from {sid}: {data}', args)\n\n async def event_stop_typing(self, sid: str, data: dict, *args):\n print(f'StopTyping event from {sid}: {data}', args)\n```\n\n\n### 2. Use with lifespan (recommended)\n\n```python\n# core/db.py\nfrom sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine\n\nengine = create_async_engine(url=\"database_uri\", echo=True)\nasync_session = async_sessionmaker(bind=engine, expire_on_commit=False, autocommit=False, autoflush=False)\n```\n\n```python\n# core/lifespan.py\nfrom contextlib import asynccontextmanager\nfrom typing import TYPE_CHECKING\n\nfrom socketio_handler import SocketManager\n\nfrom .db import async_session\n\nimport app.chat_handler # \ud83d\udc48 force-import handlers to trigger decorator registration\n\nif TYPE_CHECKING:\n from fastapi import FastAPI\n \n@asynccontextmanager\nasync def lifespan(app: \"FastAPI\"):\n \"\"\"\n Lifespan context manager for FastAPI application.\n Useful for initializing and cleaning up resources.\n \"\"\"\n async with (\n SocketManager(\n redis_url=\"redis://localhost:6379\",\n async_session=async_session,\n ) as socket_manager,\n ):\n socket_manager.mount_to_app(app)\n socket_manager.register_events()\n app.state.socket_manager = socket_manager\n yield\n```\n\n```python\n# main.py\nfrom fastapi import FastAPI\nfrom core.lifespan import lifespan\n\napp = FastAPI(lifespan=lifespan)\n```\n\n\n### 3. Connect from frontend\n```javascript\nconst socket = io('http://localhost:8000/chat', {\n auth: {\n token: 'your-auth-token'\n }\n});\n\nsocket.emit(\"typing\", { chatId: \"...\" });\n```\n\n\n## \ud83e\uddea Testing\n\nThis package includes pytest-based test utilities. \nYou can run the tests with:\n\n* Install dependencies for testing\n```shell\npoetry install --with dev\n```\n\n* Run tests\n```shell\npytest\n```\n",
"bugtrack_url": null,
"license": "MIT",
"summary": "Socket.IO wrapper for FastApi applications",
"version": "0.1.2",
"project_urls": {
"Homepage": "https://github.com/bandirom/fastapi-socketio-handler",
"Repository": "https://github.com/bandirom/fastapi-socketio-handler"
},
"split_keywords": [
"socketio",
" asyncio",
" websockets",
" python-socketio",
" asyncio-socketio",
" websockets-handler",
" fastapi"
],
"urls": [
{
"comment_text": "",
"digests": {
"blake2b_256": "f0f6d43447608d8b58c19604b3ddf89ffeea4703addaca071f1cedc88a77f8c3",
"md5": "6dd28078e152a0c6a9e36ee08456a5d9",
"sha256": "afc59f605c7f0fa6163e1fd019e629e05a0ea3f930f81ee04ffb7325d5011516"
},
"downloads": -1,
"filename": "fastapi_socketio_handler-0.1.2-py3-none-any.whl",
"has_sig": false,
"md5_digest": "6dd28078e152a0c6a9e36ee08456a5d9",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": ">=3.9",
"size": 6314,
"upload_time": "2025-08-03T11:38:36",
"upload_time_iso_8601": "2025-08-03T11:38:36.429118Z",
"url": "https://files.pythonhosted.org/packages/f0/f6/d43447608d8b58c19604b3ddf89ffeea4703addaca071f1cedc88a77f8c3/fastapi_socketio_handler-0.1.2-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": "",
"digests": {
"blake2b_256": "4a449885892ba8be84c59691cb87880b5eae6acf14920f688c8cf123158d705a",
"md5": "2ac43f98647026d1c678caf012d186d4",
"sha256": "2393215ad4bc13bdd2ddcd4464af9e847e1dda826161210845ba71c3895b2a70"
},
"downloads": -1,
"filename": "fastapi_socketio_handler-0.1.2.tar.gz",
"has_sig": false,
"md5_digest": "2ac43f98647026d1c678caf012d186d4",
"packagetype": "sdist",
"python_version": "source",
"requires_python": ">=3.9",
"size": 5219,
"upload_time": "2025-08-03T11:38:37",
"upload_time_iso_8601": "2025-08-03T11:38:37.542292Z",
"url": "https://files.pythonhosted.org/packages/4a/44/9885892ba8be84c59691cb87880b5eae6acf14920f688c8cf123158d705a/fastapi_socketio_handler-0.1.2.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2025-08-03 11:38:37",
"github": true,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"github_user": "bandirom",
"github_project": "fastapi-socketio-handler",
"travis_ci": false,
"coveralls": false,
"github_actions": true,
"lcname": "fastapi-socketio-handler"
}
JSON
