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?
The documents seem to hint that you can only use Depends
for request functions.
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