app/monitoring/metrics.py

# app/monitoring/metrics.py
import time
from typing import Dict, Any
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.requests import Request
from starlette.responses import Response


# =========================
# Simple Metrics Storage
# =========================
_metrics: Dict[str, Any] = {
    "total_requests": 0,
    "requests_by_endpoint": {},
    "requests_by_status": {},
    "total_latency_ms": 0.0,
    "start_time": time.time(),
}


# =========================
# Metrics Middleware
# =========================
class MetricsMiddleware(BaseHTTPMiddleware):
    """Middleware to collect request metrics"""
    
    async def dispatch(self, request: Request, call_next):
        start_time = time.time()
        
        # Process request
        response = await call_next(request)
        
        # Calculate latency
        process_time = (time.time() - start_time) * 1000  # ms
        
        # Update metrics
        _metrics["total_requests"] += 1
        _metrics["total_latency_ms"] += process_time
        
        endpoint = request.url.path
        _metrics["requests_by_endpoint"][endpoint] = (
            _metrics["requests_by_endpoint"].get(endpoint, 0) + 1
        )
        
        status = str(response.status_code)
        _metrics["requests_by_status"][status] = (
            _metrics["requests_by_status"].get(status, 0) + 1
        )
        
        return response


# =========================
# Metrics Endpoint
# =========================
def metrics_endpoint(request: Request) -> Dict[str, Any]:
    """
    Return metrics in JSON format.
    Compatible with Prometheus text format if needed.
    """
    uptime = time.time() - _metrics["start_time"]
    avg_latency = (
        _metrics["total_latency_ms"] / _metrics["total_requests"]
        if _metrics["total_requests"] > 0
        else 0.0
    )
    
    return {
        "total_requests": _metrics["total_requests"],
        "requests_by_endpoint": _metrics["requests_by_endpoint"],
        "requests_by_status": _metrics["requests_by_status"],
        "average_latency_ms": round(avg_latency, 2),
        "uptime_seconds": round(uptime, 2),
    }