Search code examples
pythondependency-injectionwebsocketweb-frameworksfastapi

How to use FastAPI Depends for endpoint/route in separate file?


I have an Websocket endpoint defined in separate file, like:

from starlette.endpoints import WebSocketEndpoint
from connection_service import ConnectionService


class WSEndpoint(WebSocketEndpoint):
    """Handles Websocket connections"""

    async def on_connect(self,
            websocket: WebSocket,
            connectionService: ConnectionService = Depends(ConnectionService)):
        """Handles new connection"""
        self.connectionService = connectionService
        ...

and in the main.py I register endpoint as:

from fastapi import FastAPI
from starlette.routing import WebSocketRoute
from ws_endpoint import WSEndpoint

app = FastAPI(routes=[ WebSocketRoute("/ws", WSEndpoint) ])

But Depends for my endpoint is never resolved. Is there a way to make it work?

Plus, what is even the purpose of this mechanism in FastAPI? Cannot we just use local/global variables?


Solution

  • TL;DR

    The documents seem to hint that you can only use Depends for request functions.

    Explanation

    I found a related issue #2057 in the FastAPI repo and it seems the Depends(...) only works with the requests and not anything else.

    I confirmed this by,

    from fastapi import Depends, FastAPI
    
    app = FastAPI()
    
    
    async def foo_func():
        return "This is from foo"
    
    
    async def test_depends(foo: str = Depends(foo_func)):
        return foo
    
    
    @app.get("/")
    async def read_items():
        depends_result = await test_depends()
        return depends_result

    In this case, the dependency didn't get resolved.


    Coming to your case, you can resolve the dependency something like this,

    from starlette.endpoints import WebSocketEndpoint
    from connection_service import ConnectionService
    
    
    class WSEndpoint(WebSocketEndpoint):
        async def on_connect(
                self,
                websocket: WebSocket,
                connectionService=None
        ):
            if connectionService is None:
                connectionService = ConnectionService()  # calling the depend function
    
            self.connectionService = connectionService