Skip to content

metrics

stac_fastapi.api.metrics

Prometheus metrics with low-cardinality STAC operation labels.

instrument_app 

instrument_app(app: FastAPI, endpoint: str) -> None

Instrument a FastAPI app and expose Prometheus metrics.

Must be called during app construction (e.g. from StacApi), before any requests. Adding middleware after the app has started is not supported.

Raises:

  • ImportError –

    If stac-fastapi-api[metrics] is not installed.

Source code in stac_fastapi/api/stac_fastapi/api/metrics.py 
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
def instrument_app(app: FastAPI, endpoint: str) -> None:
    """Instrument a FastAPI app and expose Prometheus metrics.

    Must be called during app construction (e.g. from ``StacApi``), before any
    requests. Adding middleware after the app has started is not supported.

    Raises:
        ImportError: If ``stac-fastapi-api[metrics]`` is not installed.
    """
    if Instrumentator is None:
        raise ImportError(
            "Prometheus metrics require the optional dependency "
            "`stac-fastapi-api[metrics]`. Install with: "
            "pip install 'stac-fastapi-api[metrics]'"
        )

    (
        Instrumentator(
            should_group_status_codes=True,
            should_ignore_untemplated=True,
            excluded_handlers=[".*/_mgmt/.*"],
        )
        .add(record_stac_metrics)
        .instrument(app)
        .expose(app, endpoint=endpoint, include_in_schema=False)
    )

record_stac_metrics 

record_stac_metrics(info: Info) -> None

Record request count and latency using STAC operation labels.

Source code in stac_fastapi/api/stac_fastapi/api/metrics.py 
 95
 96
 97
 98
 99
100
101
102
103
def record_stac_metrics(info: Info) -> None:
    """Record request count and latency using STAC operation labels."""
    route = info.request.scope.get("route")
    route_path = getattr(route, "path", None)
    router_prefix = getattr(info.request.app.state, "router_prefix", "") or ""
    operation = resolve_operation(info.method, route_path, router_prefix)

    REQUESTS.labels(operation, info.method, info.modified_status).inc()
    LATENCY.labels(operation, info.method).observe(info.modified_duration)

register_operations 

register_operations(mapping: dict[tuple[str, str], str]) -> None

Register or override operation labels for route templates.

Source code in stac_fastapi/api/stac_fastapi/api/metrics.py 
68
69
70
def register_operations(mapping: dict[tuple[str, str], str]) -> None:
    """Register or override operation labels for route templates."""
    OPERATIONS.update(mapping)

resolve_operation 

resolve_operation(method: str, route: str | None, router_prefix: str = '') -> str

Map a request method and route template to a STAC operation label.

Source code in stac_fastapi/api/stac_fastapi/api/metrics.py 
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
def resolve_operation(method: str, route: str | None, router_prefix: str = "") -> str:
    """Map a request method and route template to a STAC operation label."""
    if not route or route == "none":
        return "unknown"

    path = route
    prefix = (router_prefix or "").rstrip("/")
    if prefix and path.startswith(prefix):
        path = path[len(prefix) :] or "/"

    operation = OPERATIONS.get((method.upper(), path))
    if operation:
        return operation

    if path.startswith("/catalogs"):
        return "catalog"
    if "bulk" in path:
        return "bulk"

    return "unknown"