refactor(datadog): type metric series as discriminated TypedDicts

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

litellm/integrations/datadog/datadog_metrics.py

@@ -3,6 +3,7 @@ import gzip
 import os
 import time
 import zlib
+from collections.abc import Sequence
 from datetime import datetime
 from typing import Final, Literal
 
@@ -108,7 +109,7 @@ class DatadogMetricsLogger(CustomBatchLogger):
 
         return tags
 
-    def _add_latency_metric(self, metric: str, seconds: float, timestamp: int, tags: list[str]) -> None:
+    def _add_latency_metric(self, metric: str, seconds: float, timestamp: int, tags: Sequence[str]) -> None:
         """
         Queues a latency sample as a gauge (legacy metric name) and as a distribution
         (`<metric>.distribution`) so Datadog computes percentiles over every request
@@ -159,7 +160,9 @@ class DatadogMetricsLogger(CustomBatchLogger):
         litellm_overhead_time_ms: Final = hidden_params.get("litellm_overhead_time_ms")
         if litellm_overhead_time_ms is not None:
             overhead_tags: Final = self._extract_tags(log)  # no status_code on latency metric
-            self._add_latency_metric("litellm.overhead.latency", litellm_overhead_time_ms / 1000, timestamp, overhead_tags)
+            self._add_latency_metric(
+                "litellm.overhead.latency", litellm_overhead_time_ms / 1000, timestamp, overhead_tags
+            )
 
         # 4. Request Count / Status Code
         series_count: Final[DatadogMetricSeries] = {
@@ -212,8 +215,8 @@ class DatadogMetricsLogger(CustomBatchLogger):
         if not self.log_queue:
             return
 
-        batch: Final = tuple(self.log_queue)
-        series: Final[list[DatadogMetricSeries]] = [s for s in batch if s["type"] != "distribution"]
+        batch: Final[tuple[DatadogMetricSeries | DatadogDistributionSeries, ...]] = tuple(self.log_queue)
+        series: Final[tuple[DatadogMetricSeries, ...]] = tuple(s for s in batch if s["type"] != "distribution")
         distributions: Final[tuple[DatadogDistributionSeries, ...]] = tuple(
             s for s in batch if s["type"] == "distribution"
         )

litellm/types/integrations/datadog_metrics.py

@@ -1,6 +1,7 @@
+from collections.abc import Sequence
 from typing import Literal
 
-from typing_extensions import ReadOnly, TypedDict
+from typing_extensions import NotRequired, ReadOnly, TypedDict
 
 
 class DatadogMetricPoint(TypedDict):
@@ -8,16 +9,16 @@ class DatadogMetricPoint(TypedDict):
     value: float  # The metric value
 
 
-class DatadogMetricSeries(TypedDict, total=False):
-    metric: str
-    type: int  # 0=unspecified, 1=count, 2=rate, 3=gauge
-    points: list[DatadogMetricPoint]
-    tags: list[str]
-    interval: int | None  # Required for count (type=1) and rate (type=2) metrics
+class DatadogMetricSeries(TypedDict):
+    metric: ReadOnly[str]
+    type: ReadOnly[Literal[0, 1, 2, 3]]  # 0=unspecified, 1=count, 2=rate, 3=gauge
+    points: ReadOnly[Sequence[DatadogMetricPoint]]
+    tags: ReadOnly[Sequence[str]]
+    interval: ReadOnly[NotRequired[int]]  # Required for count (type=1) and rate (type=2) metrics
 
 
 class DatadogMetricsPayload(TypedDict):
-    series: list[DatadogMetricSeries]
+    series: ReadOnly[Sequence[DatadogMetricSeries]]
 
 
 DatadogDistributionPoint = tuple[int, tuple[float, ...]]

tests/test_litellm/integrations/datadog/test_datadog_metrics.py

@@ -235,9 +235,9 @@ async def test_overhead_latency_metric_emitted(clean_env):
     metrics = {s["metric"]: s for s in logger.log_queue}
 
     # Overhead metric must be present
-    assert (
-        "litellm.overhead.latency" in metrics
-    ), f"Expected 'litellm.overhead.latency' in emitted metrics, got: {list(metrics.keys())}"
+    assert "litellm.overhead.latency" in metrics, (
+        f"Expected 'litellm.overhead.latency' in emitted metrics, got: {list(metrics.keys())}"
+    )
     overhead = metrics["litellm.overhead.latency"]
     assert overhead["type"] == 3  # gauge
     # 250 ms → 0.25 s
@@ -380,9 +380,7 @@ async def test_async_send_batch(clean_env):
     logger = DatadogMetricsLogger(start_periodic_flush=False)
     logger.async_client = AsyncMock()
     mock_request = Request("POST", "https://api.test.datadoghq.com/api/v2/series")
-    logger.async_client.post.return_value = Response(
-        202, json={"status": "ok"}, request=mock_request
-    )
+    logger.async_client.post.return_value = Response(202, json={"status": "ok"}, request=mock_request)
 
     # Manually add a metric series to the queue
     logger.log_queue = [