Aleksei Sokol
53f14a8624
All checks were successful
Run linters on applied template / Python 3.13 lint and build (push) Successful in 1m40s
Changes: - put ObservabilityMiddleware before ExceptionHandlerMiddleware to avoid repetative code - add application startup and last metrics update metrics along with CPU usage metric and threads count - move host and port to new uvicorn section at config along with new reload and forwarded_allow_ips - add request_id and remove trace_id/span_id generation if tracing is disabled - move logging logic from utils to observability - pass trace_id/span_id in HEX form
41 lines
1.5 KiB
Python
41 lines
1.5 KiB
Python
"""structlog BoundLogger dependency functions are defined here."""
|
|
|
|
from fastapi import FastAPI, Request
|
|
from structlog.stdlib import BoundLogger
|
|
|
|
|
|
def init_dispencer(app: FastAPI, logger: BoundLogger) -> None:
|
|
"""Initialize BoundLogger dispencer at app's state."""
|
|
if hasattr(app.state, "logger"):
|
|
if not isinstance(app.state.logger_dep, BoundLogger):
|
|
raise ValueError(
|
|
f"logger attribute of app's state is already set with other value ({app.state.logger_dep})"
|
|
)
|
|
return
|
|
|
|
app.state.logger_dep = logger
|
|
|
|
|
|
def attach_to_request(request: Request, logger: BoundLogger) -> None:
|
|
"""Set logger for a concrete request. If request had already had a logger, replace it."""
|
|
if hasattr(request.state, "logger_dep"):
|
|
if not isinstance(request.state.logger_dep, BoundLogger):
|
|
logger.warning("request.state.logger is already set with other value", value=request.state.logger_dep)
|
|
request.state.logger_dep = logger
|
|
|
|
|
|
def from_app(app: FastAPI) -> BoundLogger:
|
|
"""Get a logger from app state."""
|
|
if not hasattr(app.state, "logger_dep"):
|
|
raise ValueError("BoundLogger dispencer was not initialized at app preparation")
|
|
return app.state.logger_dep
|
|
|
|
|
|
def from_request(request: Request) -> BoundLogger:
|
|
"""Get a logger from request or app state."""
|
|
if hasattr(request.state, "logger_dep"):
|
|
logger = request.state.logger_dep
|
|
if isinstance(logger, BoundLogger):
|
|
return logger
|
|
return from_app(request.app)
|