Bases: BaseHTTPMiddleware
Response HTTP info middleware for FastAPI.
Get HTTP info from response header and add it to request.state.http_info.
Inherits
BaseHTTPMiddleware: Base HTTP middleware from Starlette.
Source code in src/beans_logging_fastapi/middlewares.py
| class ResponseHTTPInfoMiddleware(BaseHTTPMiddleware):
"""Response HTTP info middleware for FastAPI.
Get HTTP info from response header and add it to `request.state.http_info`.
Inherits:
BaseHTTPMiddleware: Base HTTP middleware from Starlette.
"""
async def dispatch(self, request: Request, call_next: Callable) -> Response:
_http_info: dict[str, Any] = {}
_start_time: int = time.perf_counter_ns()
# Process request:
response: Response = await call_next(request)
# Response processed.
_end_time: int = time.perf_counter_ns()
_response_time: float = round((_end_time - _start_time) / 1_000_000, 1)
if hasattr(request.state, "http_info") and isinstance(
request.state.http_info, dict
):
_http_info: dict[str, Any] = request.state.http_info
_http_info["response_time"] = _response_time
if "X-Process-Time" in response.headers:
try:
_x_process_time = response.headers.get("X-Process-Time")
if _x_process_time:
_http_info["response_time"] = float(_x_process_time)
except ValueError:
logger.warning(
f"`X-Process-Time` header value '{response.headers.get('X-Process-Time')}' is invalid, "
"should be parseable to <float>!"
)
else:
response.headers["X-Process-Time"] = str(_http_info["response_time"])
if "X-Request-ID" not in response.headers:
response.headers["X-Request-ID"] = _http_info["request_id"]
if hasattr(request.state, "user_id"):
_http_info["user_id"] = str(request.state.user_id)
_http_info["status_code"] = response.status_code
_http_info["content_length"] = 0
if "Content-Length" in response.headers:
try:
_content_length = response.headers.get("Content-Length")
if _content_length:
_http_info["content_length"] = int(_content_length)
except ValueError:
logger.warning(
f"`Content-Length` header value '{response.headers.get('Content-Length')}' is invalid, "
"should be parseable to <int>!"
)
request.state.http_info = _http_info
return response
|