# 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):
async def on_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 on_typing(self, sid: str, data: dict, *args):
print(f'Typing event from {sid}: {data}', args)
async def on_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_handlers()
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": null,
"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/1f/a5/ff7be6efa1a71d0283bc2cce1aa2d4a3d1bcbbd82ab008d2b62a0507e73f/fastapi_socketio_handler-0.3.0.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 async def on_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 on_typing(self, sid: str, data: dict, *args):\n print(f'Typing event from {sid}: {data}', args)\n\n async def on_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_handlers()\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.3.0",
"project_urls": {
"changelog": "https://github.com/bandirom/fastapi-socketio-handler/blob/main/CHANGELOG.md",
"homepage": "https://github.com/bandirom/fastapi-socketio-handler",
"issues": "https://github.com/bandirom/fastapi-socketio-handler/issues",
"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": "538a0078cfe17659ebf0e6aadb1ed78d5c6ac060217faa535e9c4569dc31939d",
"md5": "69f5579c4d72ed15f280d37e18551ead",
"sha256": "bdb7a335a6a8f872ecf2fb75c76ccba3e42fb0b9aea79d77416124b7ae50e66b"
},
"downloads": -1,
"filename": "fastapi_socketio_handler-0.3.0-py3-none-any.whl",
"has_sig": false,
"md5_digest": "69f5579c4d72ed15f280d37e18551ead",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": ">=3.9",
"size": 6309,
"upload_time": "2025-08-31T11:23:02",
"upload_time_iso_8601": "2025-08-31T11:23:02.391565Z",
"url": "https://files.pythonhosted.org/packages/53/8a/0078cfe17659ebf0e6aadb1ed78d5c6ac060217faa535e9c4569dc31939d/fastapi_socketio_handler-0.3.0-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": "",
"digests": {
"blake2b_256": "1fa5ff7be6efa1a71d0283bc2cce1aa2d4a3d1bcbbd82ab008d2b62a0507e73f",
"md5": "e9d7b80ef2cc526a762df538d9d9724b",
"sha256": "da02432ce4afe0331b307f2a1d1c4a7ba179cccbf37f25adfbc6541ae7da4b35"
},
"downloads": -1,
"filename": "fastapi_socketio_handler-0.3.0.tar.gz",
"has_sig": false,
"md5_digest": "e9d7b80ef2cc526a762df538d9d9724b",
"packagetype": "sdist",
"python_version": "source",
"requires_python": ">=3.9",
"size": 5344,
"upload_time": "2025-08-31T11:23:03",
"upload_time_iso_8601": "2025-08-31T11:23:03.511820Z",
"url": "https://files.pythonhosted.org/packages/1f/a5/ff7be6efa1a71d0283bc2cce1aa2d4a3d1bcbbd82ab008d2b62a0507e73f/fastapi_socketio_handler-0.3.0.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2025-08-31 11:23:03",
"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
