I'm using python-socketio
and trying to pass the server instance to my app routers as dependency:
main.py
file:
import socketio
import uvicorn
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from src.api.rest.api_v1.route_builder import build_routes
app = FastAPI()
app.add_middleware(
CORSMiddleware,
allow_credentials=True,
allow_methods=["*"],
)
sio = socketio.AsyncServer(cors_allowed_origins="*", async_mode="asgi")
app = build_routes(app)
sio_asgi_app = socketio.ASGIApp(sio, app)
@app.get("/")
async def root():
return {"message": "Hello from main server !"}
def start_server():
api_port = "0.0.0.0"
api_host = "8000"
uvicorn.run(
"src.api.main:sio_asgi_app",
host=api_host,
port=int(api_port),
log_level="info",
reload=True,
)
if __name__ == "__main__":
start_server()
product_routes.py
file:
from typing import Annotated
import socketio
from fastapi import APIRouter, Depends
router = APIRouter(prefix="/products")
@router.get("/emit")
async def emit_message(
sio: Annotated[socketio.AsyncServer, Depends()]
) -> None:
await sio.emit("reply", {"message": "Hello from server !"})
build_routes.py
file:
def build_routes(app: FastAPI) -> FastAPI:
app.include_router(
r_products,
prefix="/v1",
dependencies=[Depends(socketio.AsyncServer)],
)
return app
So at this point I'm not sure how to pass that dependency to the routes.
When I hit the emit
endpoint I get this error response:
{
"detail": [
{
"type": "missing",
"loc": [
"query",
"kwargs"
],
"msg": "Field required",
"input": null,
"url": "https://errors.pydantic.dev/2.5/v/missing"
}
]
}
You need to create function, where just return AsyncServer object. Then use this function as a dependency in your endpoint functions. It's better to create this dependency in separate file to avoid import errors.
dependencies.py
import socketio
sio = socketio.AsyncServer(cors_allowed_origins="*", async_mode="asgi")
def sio_dep() -> socketio.AsyncServer:
return sio
then pass this function as a parameter to Depends:
from typing import Annotated
from dependencies import sio_dep
from fastapi import APIRouter, Depends
router = APIRouter(prefix="/products")
@router.get("/emit")
async def emit_message(
sio: Annotated[socketio.AsyncServer, Depends(sio_dep)]
) -> None:
await sio.emit("reply", {"message": "Hello from server !"})