MetricsMiddleware

Warning

You need to install dependencies to use The Prometheus Metrics Middleware.

$ pip install authx_extra[prometheus]

authx_extra.metrics.MetricsMiddleware

MetricsMiddleware(app, prefix='authx_', buckets=(0.002, 0.05, 0.1, prometheus_client.utils.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)

app instance-attribute

app = app

dispatch_func instance-attribute

dispatch_func = dispatch if dispatch is None else dispatch

request_count instance-attribute

request_count = request_count(prefix)

request_time instance-attribute

request_time = request_time(prefix, buckets)

dispatch async

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