websocket_session() — fastapi Function Reference

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

Function python FastAPI Routing called by 1

Entity Profile

FastAPI→ Routing→ websocket_session() — fastapi Function Reference

Dependency Diagram

graph TD
  40cf508d_1e02_35ba_ab86_a9399504216d["websocket_session()"]
  de395a51_26f8_3424_1af0_2f5bef39c893["routing.py"]
  40cf508d_1e02_35ba_ab86_a9399504216d -->|defined in| de395a51_26f8_3424_1af0_2f5bef39c893
  0df3979a_d2ac_6dd9_1bb1_6629b89a941b["__init__()"]
  0df3979a_d2ac_6dd9_1bb1_6629b89a941b -->|calls| 40cf508d_1e02_35ba_ab86_a9399504216d
  style 40cf508d_1e02_35ba_ab86_a9399504216d fill:#6366f1,stroke:#818cf8,color:#fff

Relationship Graph

Source Code

fastapi/routing.py lines 128–149

def websocket_session(
    func: Callable[[WebSocket], Awaitable[None]],
) -> ASGIApp:
    """
    Takes a coroutine `func(session)`, and returns an ASGI application.
    """
    # assert asyncio.iscoroutinefunction(func), "WebSocket endpoints must be async"

    async def app(scope: Scope, receive: Receive, send: Send) -> None:
        session = WebSocket(scope, receive=receive, send=send)

        async def app(scope: Scope, receive: Receive, send: Send) -> None:
            async with AsyncExitStack() as request_stack:
                scope["fastapi_inner_astack"] = request_stack
                async with AsyncExitStack() as function_stack:
                    scope["fastapi_function_astack"] = function_stack
                    await func(session)

        # Same as in Starlette
        await wrap_app_handling_exceptions(app, session)(scope, receive, send)

    return app