MetricsMiddleware(app, prefix='authx_', buckets=(0.002, 0.05, 0.1, INF))
Bases: BaseHTTPMiddleware
Metrics middleware collecting prometheus metrics for each request.
Initialize a new MetricsMiddleware instance.
| PARAMETER | DESCRIPTION |
app | TYPE: FastAPI |
prefix | TYPE: str DEFAULT: 'authx_' |
buckets | TYPE: Tuple[float, ...] DEFAULT: (0.002, 0.05, 0.1, INF) |
Source code in authx_extra/metrics.py
| def __init__(
self,
app: FastAPI,
prefix: str = "authx_",
buckets: typing.Tuple[float, ...] = (
0.002,
0.05,
0.1,
prometheus_client.utils.INF,
),
) -> None:
"""Initialize a new MetricsMiddleware instance."""
super().__init__(app)
self.request_count = request_count(prefix)
self.request_time = request_time(prefix, buckets)
|
dispatch_func = dispatch if dispatch is None else dispatch
dispatch(request, call_next)
Record request method, path and status when dispatching.
| PARAMETER | DESCRIPTION |
request | TYPE: Request |
call_next | TYPE: Callable |
Source code in authx_extra/metrics.py
| async def dispatch(self, request: Request, call_next: typing.Callable):
"""Record request method, path and status when dispatching."""
method = request.method
path = request.url.path
status = 500
begin = time.time()
try:
response = await call_next(request)
status = response.status_code
finally:
# track urls w/ params grouped, eg. /items/123 -> /items/{id}
router = request.scope.get("router")
endpoint = request.scope.get("endpoint")
if router and endpoint:
for route in router.routes:
route_app = getattr(route, "app", None)
route_endpoint = getattr(route, "endpoint", None)
if endpoint in (route_app, route_endpoint):
path = route.path
break
end = time.time()
labels = [method, path, status]
self.request_count.labels(*labels).inc()
self.request_time.labels(*labels).observe(end - begin)
return response
|