websocket_middleware() — fastapi Function Reference

Architecture documentation for the websocket_middleware() function in test_ws_router.py from the fastapi codebase.

Function python FastAPI Routing

Entity Profile

FastAPI→ Routing→ websocket_middleware() — fastapi Function Reference

Dependency Diagram

graph TD
  2dd33bcf_c530_dad9_7f5d_7247407fd0b2["websocket_middleware()"]
  b2e4c2fb_8042_93dc_9d79_2ae1a12773d6["test_ws_router.py"]
  2dd33bcf_c530_dad9_7f5d_7247407fd0b2 -->|defined in| b2e4c2fb_8042_93dc_9d79_2ae1a12773d6
  style 2dd33bcf_c530_dad9_7f5d_7247407fd0b2 fill:#6366f1,stroke:#818cf8,color:#fff

Relationship Graph

Source Code

tests/test_ws_router.py lines 188–207

def websocket_middleware(middleware_func):
    """
    Helper to create a Starlette pure websocket middleware
    """

    def middleware_constructor(app):
        @functools.wraps(app)
        async def wrapped_app(scope, receive, send):
            if scope["type"] != "websocket":
                return await app(scope, receive, send)  # pragma: no cover

            async def call_next():
                return await app(scope, receive, send)

            websocket = WebSocket(scope, receive=receive, send=send)
            return await middleware_func(websocket, call_next)

        return wrapped_app

    return middleware_constructor