Skip to content

fastapi_tools.middleware

Custom middleware for the webapp.

Classes:

Functions:

RequestIDMiddleware

Bases: BaseHTTPMiddleware

Attach a unique request ID to every request.

Methods:

  • dispatch –

    Add X-Request-ID header to request and response.

dispatch async 

dispatch(request: Request, call_next: Callable) -> Response

Add X-Request-ID header to request and response.

Source code in src/fastapi_tools/middleware.py 
25
26
27
28
29
30
31
async def dispatch(self, request: Request, call_next: Callable) -> Response:
    """Add X-Request-ID header to request and response."""
    request_id = str(uuid.uuid4())
    request.state.request_id = request_id
    response = await call_next(request)
    response.headers["X-Request-ID"] = request_id
    return response

RequestLoggingMiddleware

Bases: BaseHTTPMiddleware

Log incoming requests and their response status/duration.

Methods:

  • dispatch –

    Log request and response details.

dispatch async 

dispatch(request: Request, call_next: Callable) -> Response

Log request and response details.

Source code in src/fastapi_tools/middleware.py 
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
async def dispatch(self, request: Request, call_next: Callable) -> Response:
    """Log request and response details."""
    start_time = time.perf_counter()
    request_id = getattr(request.state, "request_id", "unknown")
    lg.info(
        f"[{request_id}] {request.method} {request.url.path} "
        f"from {request.client.host if request.client else 'unknown'}",
    )
    response = await call_next(request)
    duration_ms = (time.perf_counter() - start_time) * 1000
    lg.info(
        f"[{request_id}] {request.method} {request.url.path} "
        f"-> {response.status_code} ({duration_ms:.2f}ms)",
    )
    return response

SecurityHeadersMiddleware 

SecurityHeadersMiddleware(
    app: ASGIApp, *, is_production: bool = False
)

Bases: BaseHTTPMiddleware

Add security-related HTTP headers to every response.

Methods:

  • dispatch –

    Add security headers to the response.

Source code in src/fastapi_tools/middleware.py 
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
def __init__(self, app: ASGIApp, *, is_production: bool = False) -> None:  # noqa: D107
    super().__init__(app)
    self.is_production = is_production
    self.strict_csp = (
        "default-src 'self'"
        "; script-src 'self'"
        "; style-src 'self' 'unsafe-inline'"
        "; img-src 'self' data: https://lh3.googleusercontent.com"
        "; font-src 'self'"
        "; connect-src 'self'"
        "; frame-ancestors 'none'"
        "; base-uri 'self'"
        "; form-action 'self'"
        "; object-src 'none'"
        "; worker-src 'self'"
        "; manifest-src 'self'"
    )
    self.docs_csp = (
        "default-src 'self'"
        "; script-src 'self' 'unsafe-inline'"
        "; style-src 'self' 'unsafe-inline'"
        "; img-src 'self' data:"
        "; font-src 'self'"
        "; connect-src 'self'"
        "; frame-ancestors 'none'"
        "; base-uri 'self'"
        "; form-action 'self'"
        "; object-src 'none'"
        "; worker-src 'self'"
        "; manifest-src 'self'"
    )

dispatch async 

dispatch(request: Request, call_next: Callable) -> Response

Add security headers to the response.

Source code in src/fastapi_tools/middleware.py 
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
async def dispatch(self, request: Request, call_next: Callable) -> Response:
    """Add security headers to the response."""
    response = await call_next(request)
    response.headers["X-Content-Type-Options"] = "nosniff"
    response.headers["X-Frame-Options"] = "DENY"
    response.headers["X-XSS-Protection"] = "1; mode=block"
    response.headers["Referrer-Policy"] = "strict-origin-when-cross-origin"

    path = request.url.path
    if any(path.startswith(prefix) for prefix in self._DOCS_PREFIXES):
        response.headers["Content-Security-Policy"] = self.docs_csp
    else:
        response.headers["Content-Security-Policy"] = self.strict_csp

    if self.is_production:
        response.headers["Strict-Transport-Security"] = (
            "max-age=31536000; includeSubDomains"
        )
    return response

setup_middleware 

setup_middleware(
    app: ASGIApp, config: WebappConfig
) -> None

Add custom middleware stack to a FastAPI application.

Parameters:

  • app (ASGIApp) –

    FastAPI application instance.

  • config (WebappConfig) –

    Webapp configuration.

Source code in src/fastapi_tools/middleware.py 
112
113
114
115
116
117
118
119
120
121
122
123
124
125
def setup_middleware(app: ASGIApp, config: WebappConfig) -> None:
    """Add custom middleware stack to a FastAPI application.

    Args:
        app: FastAPI application instance.
        config: Webapp configuration.
    """
    from fastapi import FastAPI  # noqa: PLC0415

    if not isinstance(app, FastAPI):
        return
    app.add_middleware(RequestLoggingMiddleware)
    app.add_middleware(SecurityHeadersMiddleware, is_production=not config.debug)
    app.add_middleware(RequestIDMiddleware)