Real-time Communication with WebSockets

WebSockets in this codebase are integrated into the routing system through the APIRouter class and the specialized APIWebSocketRoute class. This integration allows WebSocket endpoints to benefit from the same features as standard HTTP routes, including path parameters, query parameters, and dependency injection.

Defining WebSocket Endpoints

The primary way to define a WebSocket endpoint is using the @router.websocket() decorator provided by APIRouter. This decorator registers the endpoint as an APIWebSocketRoute.

from fastapi import APIRouter, WebSocket

router = APIRouter()

@router.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()
    while True:
        data = await websocket.receive_text()
        await websocket.send_text(f"Message text was: {data}")

Internally, APIRouter.websocket calls add_api_websocket_route, which instantiates an APIWebSocketRoute and appends it to the router's list of routes.

Path and Query Parameters

WebSocket routes support dynamic path segments and query parameters. These are automatically extracted and passed to the endpoint function based on its signature.

As seen in tests/test_ws_router.py, you can define complex paths and expect query parameters:

@router.websocket("/router/{pathparam:path}")
async def routerindexparams(websocket: WebSocket, pathparam: str, queryparam: str):
    await websocket.accept()
    await websocket.send_text(pathparam)
    await websocket.send_text(queryparam)
    await websocket.close()

In this example, if a client connects to /router/some/path?queryparam=value, the pathparam argument will receive "some/path" and queryparam will receive "value".

Dependency Injection in WebSockets

One of the most powerful features of APIWebSocketRoute is its support for dependency injection. Unlike standard Starlette WebSockets, FastAPI's implementation resolves dependencies before the endpoint is executed.

Endpoint-level Dependencies

You can use the Depends function directly in the WebSocket endpoint signature:

async def ws_dependency():
    return "Socket Dependency"

@router.websocket("/router-ws-depends/")
async def router_ws_decorator_depends(
    websocket: WebSocket, data=Depends(ws_dependency)
):
    await websocket.accept()
    await websocket.send_text(data)
    await websocket.close()

Router-level Dependencies

Dependencies can also be applied to all WebSocket routes within an APIRouter by passing them to the APIRouter constructor or when including the router:

# All routes in this router will execute 'global_dependency'
router = APIRouter(dependencies=[Depends(global_dependency)])

The APIWebSocketRoute Implementation

The APIWebSocketRoute class (found in fastapi/routing.py) is responsible for bridging the gap between the raw WebSocket connection and FastAPI's dependency system.

Connection Lifecycle and Dependencies

When a client connects, APIWebSocketRoute does not call the endpoint directly. Instead, it uses get_websocket_app to create an internal ASGI application that:

1. Solves all required dependencies using solve_dependencies.
2. Handles validation of parameters and dependencies.
3. Executes the actual endpoint function.

The class signature shows how it stores these dependencies:

class APIWebSocketRoute(routing.WebSocketRoute):
    def __init__(
        self,
        path: str,
        endpoint: Callable[..., Any],
        *,
        name: str | None = None,
        dependencies: Sequence[params.Depends] | None = None,
        dependency_overrides_provider: Any | None = None,
    ) -> None:
        # ... initialization logic ...
        self.dependant = get_dependant(
            path=self.path_format, call=self.endpoint, scope="function"
        )
        # ... wraps the endpoint in a session handler ...
        self.app = websocket_session(
            get_websocket_app(
                dependant=self.dependant,
                dependency_overrides_provider=dependency_overrides_provider,
                embed_body_fields=self._embed_body_fields,
            )
        )

Validation Errors

If a dependency or parameter fails validation (e.g., a required header is missing), FastAPI raises a WebSocketRequestValidationError. By default, this results in the WebSocket connection being closed with the code 1008 (WS_1008_POLICY_VIOLATION), as demonstrated in the project's test suite (tests/test_ws_router.py).

Router Integration

When an APIRouter is included in another router or the main FastAPI app via include_router, APIWebSocketRoute instances are correctly migrated with their prefixes and dependencies.

In APIRouter.include_router, the logic specifically checks for APIWebSocketRoute:

elif isinstance(route, APIWebSocketRoute):
    current_dependencies = []
    if dependencies:
        current_dependencies.extend(dependencies)
    if route.dependencies:
        current_dependencies.extend(route.dependencies)
    self.add_api_websocket_route(
        prefix + route.path,
        route.endpoint,
        dependencies=current_dependencies,
        name=route.name,
    )

This ensures that a WebSocket defined at @router.websocket("/ws") inside a router with prefix /api correctly resolves to /api/ws when the router is included.