25 lines
943 B
Python
25 lines
943 B
Python
"""Prometheus metrics endpoint."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from fastapi import APIRouter, Header, HTTPException, Response, status
|
|
from prometheus_client import CONTENT_TYPE_LATEST, generate_latest
|
|
|
|
from src.contract_check.core.config import get_settings
|
|
|
|
router = APIRouter(tags=["metrics"])
|
|
|
|
|
|
@router.get("/metrics")
|
|
async def metrics(authorization: str | None = Header(default=None)) -> Response:
|
|
token = get_settings().metrics_bearer_token
|
|
if token:
|
|
if not authorization or not authorization.lower().startswith("bearer "):
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED, detail="missing bearer token"
|
|
)
|
|
if authorization[7:].strip() != token:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED, detail="invalid bearer token"
|
|
)
|
|
return Response(generate_latest(), media_type=CONTENT_TYPE_LATEST)
|