refactor: type request/provider logging payloads and monitoring helpers

Author: Automaticare

app/logging.py

@@ -8,7 +8,7 @@
 import uuid
 from collections.abc import Iterable
 from datetime import UTC, datetime
-from typing import Any
+from typing import Any, TypedDict
 
 from flask import g, has_request_context, request
 from werkzeug.exceptions import HTTPException
@@ -43,6 +43,32 @@
 }
 
 
+class RequestLogPayload(TypedDict, total=False):
+    event: str
+    route: str
+    method: str
+    status: int
+    duration_ms: float
+    request_id: str
+    path: str
+    source: str
+    stale: bool
+    error: str
+    client_ip: str
+
+
+class ProviderLogPayload(TypedDict, total=False):
+    event: str
+    provider: str
+    base: str
+    status: str
+    duration_ms: float
+    request_id: str | None
+    source: str
+    stale: bool
+    error: str
+
+
 class JSONLogFormatter(logging.Formatter):
     """Format LogRecord instances into structured JSON strings."""
 
@@ -170,25 +196,27 @@ def _build_request_log_extra(
     request_id: str | None,
     duration_ms: float | None,
     error: str | None,
-) -> dict[str, Any]:
+) -> RequestLogPayload:
     route = request.url_rule.rule if request.url_rule else request.path
-    payload: dict[str, Any] = {
+    payload: RequestLogPayload = {
         "event": event,
         "route": route,
         "method": request.method,
         "status": status,
-        "duration_ms": round(duration_ms, 3) if duration_ms is not None else None,
-        "request_id": request_id,
         "path": request.path,
         "source": "api",
         "stale": False,
     }
+    if duration_ms is not None:
+        payload["duration_ms"] = round(duration_ms, 3)
+    if request_id:
+        payload["request_id"] = request_id
     if error:
         payload["error"] = error
     if request.remote_addr:
         payload["client_ip"] = request.remote_addr
 
-    return {key: value for key, value in payload.items() if value is not None}
+    return payload
 
 
 def _extract_extras(record_dict: dict[str, Any]) -> dict[str, Any]:
@@ -243,20 +271,21 @@ def provider_log_extra(
     duration_ms: float | None,
     stale: bool,
     error: str | None = None,
-) -> dict[str, Any]:
-    payload: dict[str, Any] = {
+) -> ProviderLogPayload:
+    payload: ProviderLogPayload = {
         "event": event,
         "provider": provider,
         "base": base,
         "status": status,
-        "duration_ms": round(duration_ms, 3) if duration_ms is not None else None,
-        "request_id": _current_request_id(),
         "source": provider,
         "stale": stale,
     }
+    if duration_ms is not None:
+        payload["duration_ms"] = round(duration_ms, 3)
+    payload["request_id"] = _current_request_id()
     if error:
         payload["error"] = error
-    return {key: value for key, value in payload.items() if value is not None}
+    return payload
 
 
 def _current_request_id() -> str | None:

app/monitoring.py

@@ -6,23 +6,28 @@
 from collections.abc import Callable, Iterator, Mapping, MutableMapping
 from contextlib import contextmanager
 from time import perf_counter
-from typing import Any, cast
+from typing import Any, Protocol, cast
 
 from flask import Flask, current_app, g
 
+
+class MetadataFactory(Protocol):
+    def __call__(self, *args: Any, **kwargs: Any) -> Mapping[str, Any] | None: ...
+
+
 LOG_EVENT_NAME = "performance.timing"
 CONFIG_ENABLED_KEY = "TIMING_LOGS_ENABLED"
 CONFIG_THRESHOLD_KEY = "TIMING_MIN_DURATION_MS"
 
 
-def _is_enabled(app) -> bool:
+def _is_enabled(app: Flask) -> bool:
     value = app.config.get(CONFIG_ENABLED_KEY, False)
     if isinstance(value, bool):
         return value
     return str(value).strip().lower() in {"1", "true", "yes", "on"}
 
 
-def _threshold_ms(app) -> float | None:
+def _threshold_ms(app: Flask) -> float | None:
     value = app.config.get(CONFIG_THRESHOLD_KEY)
     if value is None:
         return None
@@ -112,7 +117,7 @@ def timed_operation(
 def timed(
     event: str,
     *,
-    metadata_factory: Callable[..., Mapping[str, Any] | None] | None = None,
+    metadata_factory: MetadataFactory | None = None,
 ) -> Callable[[Callable[..., Any]], Callable[..., Any]]:
     """Decorator variant of ``timed_operation`` for function instrumentation."""