After hours learning playing around with Dependency Injection and routes/endpoints in FastAPI here is what I found.
Route vs Endpoint
First of all want to point out that Endpoint is a concept that exists in Starlette and no in FastAPI. In my question I show code where I use WebSocketEndpoint class and Dependency Injection will not work in FastAPI. Read further to understand why.
Dependency injection (DI)
DI in FastAPI is not a classic pattern that we know, it is not resolving magically all dependencies everywhere.
Depends is only resolved for FastAPI routes, meaning using methods: add_api_route and add_api_websocket_route, or their decorator analogs: api_route and websocket, which are just wrappers around first two.
Then dependencies are going to be resolved when request comes to the route by FastAPI. This is important to understand that FastAPI is resolving dependencies and not Starlette. FastAPI is build on top of Starlette and you may want to use also some "raw" Starlette features, like: add_route or add_websocket_route, but then you will not have Depends resolution for those.
Also, DI in FastAPI can be used to resolve instances of classes but it's not its main purpose + it makes no sense in Python because you can just use CLOSURE. Where Depends shine is when you need some sort of request validation (what Django accomplishes with decorators). In this usage Depends is great, because it resolves route dependencies and those sub dependencies. Check out my code below and I use auth_check.
Code example
As a bonus I want to have websocket route as a class in separate file with separated methods for connect, disconnect and receive. Also, I want to have authentication check in separate file to be able to swap it in easily.
# main.py
from fastapi import FastAPI
from ws_route import WSRoute
app = FastAPI()
app.add_api_websocket_route("/ws", WSRoute)
# auth.py
from fastapi import WebSocket
def auth_check(websocket: WebSocket):
# `websocket` instance is resolved automatically
# and other `Depends` as well. They are what's called sub dependencies.
# Implement your authentication logic here:
# Parse Headers or query parameters (which is usually a way for websockets)
# and perform verification
return True
# ws_route.py
import typing
import starlette.status as status
from fastapi import WebSocket, WebSocketDisconnect, Depends
from auth import auth_check
class WSRoute:
def __init__(self,
websocket: WebSocket,
is_authenticated: bool = Depends(auth_check)):
self._websocket = websocket
def __await__(self) -> typing.Generator:
return self.dispatch().__await__()
async def dispatch(self) -> None:
# Websocket lifecycle
await self._on_connect()
close_code: int = status.WS_1000_NORMAL_CLOSURE
try:
while True:
data = await self._websocket.receive_text()
await self._on_receive(data)
except WebSocketDisconnect:
# Handle client normal disconnect here
pass
except Exception as exc:
# Handle other types of errors here
close_code = status.WS_1011_INTERNAL_ERROR
raise exc from None
finally:
await self._on_disconnect(close_code)
async def _on_connect(self):
# Handle your new connection here
await self._websocket.accept()
pass
async def _on_disconnect(self, close_code: int):
# Handle client disconnect here
pass
async def _on_receive(self, msg: typing.Any):
# Handle client messaging here
pass
