pythonfastapi

Pass SocketIO server instance to FastAPI routers as dependency


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"
        }
    ]
}

Solution

  • 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 !"})