Skip to content

stac_auth_proxy.metrics

Optional Prometheus metrics for STAC operations.

classify_operation(method: str, path: str) -> str

Map a request to a low-cardinality STAC operation name.

Source code in src/stac_auth_proxy/metrics.py 
39
40
41
42
43
44
def classify_operation(method: str, path: str) -> str:
    """Map a request to a low-cardinality STAC operation name."""
    for pattern, methods in _COMPILED:
        if pattern.match(path):
            return methods.get(method.upper(), "unknown")
    return "unknown"

instrument_app(app: Any, excluded_handlers: Optional[Sequence[str]] = None) -> None

Instrument a FastAPI app and expose Prometheus metrics when available.

Source code in src/stac_auth_proxy/metrics.py 
47
48
49
50
51
52
53
54
def instrument_app(
    app: Any,
    excluded_handlers: Optional[Sequence[str]] = None,
) -> None:
    """Instrument a FastAPI app and expose Prometheus metrics when available."""
    if not METRICS_AVAILABLE:
        return
    _instrument_app(app, list(excluded_handlers or []))

record_stac_metrics(info: Info) -> None

Record request count and latency using STAC operation labels.

Source code in src/stac_auth_proxy/metrics.py 
74
75
76
77
78
79
80
81
def record_stac_metrics(info: Info) -> None:
    """Record request count and latency using STAC operation labels."""
    # The routed path: the restored root path isn't part of STAC operations
    operation = classify_operation(
        info.request.method, checked_path(info.request.scope)
    )
    REQUESTS.labels(operation, info.method, info.modified_status).inc()
    LATENCY.labels(operation, info.method).observe(info.modified_duration)