get_websocket_app() — fastapi Function Reference

Architecture documentation for the get_websocket_app() function in routing.py from the fastapi codebase.

Function python FastAPI Routing calls 3 called by 1

Entity Profile

FastAPI→ Routing→ get_websocket_app() — fastapi Function Reference

Dependency Diagram

graph TD
  a0d87950_5de5_3ab2_72f7_0a20a6e00434["get_websocket_app()"]
  de395a51_26f8_3424_1af0_2f5bef39c893["routing.py"]
  a0d87950_5de5_3ab2_72f7_0a20a6e00434 -->|defined in| de395a51_26f8_3424_1af0_2f5bef39c893
  0df3979a_d2ac_6dd9_1bb1_6629b89a941b["__init__()"]
  0df3979a_d2ac_6dd9_1bb1_6629b89a941b -->|calls| a0d87950_5de5_3ab2_72f7_0a20a6e00434
  74570640_d48f_6978_d994_fcf56400f5f3["_extract_endpoint_context()"]
  a0d87950_5de5_3ab2_72f7_0a20a6e00434 -->|calls| 74570640_d48f_6978_d994_fcf56400f5f3
  1b892f95_174f_a6fa_5ff9_9cd8b88e8ee0["get()"]
  a0d87950_5de5_3ab2_72f7_0a20a6e00434 -->|calls| 1b892f95_174f_a6fa_5ff9_9cd8b88e8ee0
  fc3220b9_b8bc_99f6_605f_911723c78183["solve_dependencies()"]
  a0d87950_5de5_3ab2_72f7_0a20a6e00434 -->|calls| fc3220b9_b8bc_99f6_605f_911723c78183
  style a0d87950_5de5_3ab2_72f7_0a20a6e00434 fill:#6366f1,stroke:#818cf8,color:#fff

Relationship Graph

Source Code

fastapi/routing.py lines 477–511

def get_websocket_app(
    dependant: Dependant,
    dependency_overrides_provider: Optional[Any] = None,
    embed_body_fields: bool = False,
) -> Callable[[WebSocket], Coroutine[Any, Any, Any]]:
    async def app(websocket: WebSocket) -> None:
        endpoint_ctx = (
            _extract_endpoint_context(dependant.call)
            if dependant.call
            else EndpointContext()
        )
        if dependant.path:
            # For mounted sub-apps, include the mount path prefix
            mount_path = websocket.scope.get("root_path", "").rstrip("/")
            endpoint_ctx["path"] = f"WS {mount_path}{dependant.path}"
        async_exit_stack = websocket.scope.get("fastapi_inner_astack")
        assert isinstance(async_exit_stack, AsyncExitStack), (
            "fastapi_inner_astack not found in request scope"
        )
        solved_result = await solve_dependencies(
            request=websocket,
            dependant=dependant,
            dependency_overrides_provider=dependency_overrides_provider,
            async_exit_stack=async_exit_stack,
            embed_body_fields=embed_body_fields,
        )
        if solved_result.errors:
            raise WebSocketRequestValidationError(
                solved_result.errors,
                endpoint_ctx=endpoint_ctx,
            )
        assert dependant.call is not None, "dependant.call must be a function"
        await dependant.call(**solved_result.values)

    return app