From 565a622cf9d67147b038b927078a1f3ac01de366 Mon Sep 17 00:00:00 2001 From: Yuta Saito Date: Thu, 8 Jan 2026 12:09:58 +0900 Subject: [PATCH] feat: scaffold Focus export logging skeleton --- litellm/integrations/focus/__init__.py | 0 litellm/integrations/focus/database.py | 22 +++ .../focus/destinations/__init__.py | 12 ++ .../integrations/focus/destinations/base.py | 30 ++++ .../focus/destinations/factory.py | 21 +++ .../focus/destinations/s3_destination.py | 32 +++++ .../integrations/focus/focus_export_logger.py | 129 ++++++++++++++++++ .../focus/serializers/__init__.py | 6 + .../integrations/focus/serializers/base.py | 18 +++ .../integrations/focus/serializers/parquet.py | 16 +++ litellm/integrations/focus/transformer.py | 13 ++ 11 files changed, 299 insertions(+) create mode 100644 litellm/integrations/focus/__init__.py create mode 100644 litellm/integrations/focus/database.py create mode 100644 litellm/integrations/focus/destinations/__init__.py create mode 100644 litellm/integrations/focus/destinations/base.py create mode 100644 litellm/integrations/focus/destinations/factory.py create mode 100644 litellm/integrations/focus/destinations/s3_destination.py create mode 100644 litellm/integrations/focus/focus_export_logger.py create mode 100644 litellm/integrations/focus/serializers/__init__.py create mode 100644 litellm/integrations/focus/serializers/base.py create mode 100644 litellm/integrations/focus/serializers/parquet.py create mode 100644 litellm/integrations/focus/transformer.py diff --git a/litellm/integrations/focus/__init__.py b/litellm/integrations/focus/__init__.py new file mode 100644 index 00000000000..e69de29bb2d diff --git a/litellm/integrations/focus/database.py b/litellm/integrations/focus/database.py new file mode 100644 index 00000000000..a23d16278af --- /dev/null +++ b/litellm/integrations/focus/database.py @@ -0,0 +1,22 @@ +"""Database access helpers for Focus export.""" + +from __future__ import annotations + +from datetime import datetime +from typing import Optional + +import polars as pl + + +class FocusLiteLLMDatabase: + """Retrieves LiteLLM usage data for Focus export workflows.""" + + async def get_usage_data( + self, + *, + limit: Optional[int] = None, + start_time_utc: Optional[datetime] = None, + end_time_utc: Optional[datetime] = None, + ) -> pl.DataFrame: + """Return usage data for the requested window.""" + raise NotImplementedError diff --git a/litellm/integrations/focus/destinations/__init__.py b/litellm/integrations/focus/destinations/__init__.py new file mode 100644 index 00000000000..233f1da0c9b --- /dev/null +++ b/litellm/integrations/focus/destinations/__init__.py @@ -0,0 +1,12 @@ +"""Destination implementations for Focus export.""" + +from .base import FocusDestination, FocusTimeWindow +from .factory import FocusDestinationFactory +from .s3_destination import FocusS3Destination + +__all__ = [ + "FocusDestination", + "FocusDestinationFactory", + "FocusTimeWindow", + "FocusS3Destination", +] diff --git a/litellm/integrations/focus/destinations/base.py b/litellm/integrations/focus/destinations/base.py new file mode 100644 index 00000000000..8042a7e23b9 --- /dev/null +++ b/litellm/integrations/focus/destinations/base.py @@ -0,0 +1,30 @@ +"""Abstract destination interfaces for Focus export.""" + +from __future__ import annotations + +from dataclasses import dataclass +from datetime import datetime +from typing import Protocol + + +@dataclass(frozen=True) +class FocusTimeWindow: + """Represents the span of data exported in a single batch.""" + + start_time: datetime + end_time: datetime + frequency: str + + +class FocusDestination(Protocol): + """Protocol for anything that can receive Focus export files.""" + + async def deliver( + self, + *, + content: bytes, + time_window: FocusTimeWindow, + filename: str, + ) -> None: + """Persist the serialized export for the provided time window.""" + ... diff --git a/litellm/integrations/focus/destinations/factory.py b/litellm/integrations/focus/destinations/factory.py new file mode 100644 index 00000000000..36163ece61c --- /dev/null +++ b/litellm/integrations/focus/destinations/factory.py @@ -0,0 +1,21 @@ +"""Factory helpers for Focus export destinations.""" + +from __future__ import annotations + +from typing import Optional + +from .base import FocusDestination + + +class FocusDestinationFactory: + """Builds destination instances based on provider/config settings.""" + + @staticmethod + def create( + *, + provider: str, + prefix: str, + config: Optional[dict] = None, + ) -> FocusDestination: + """Return a destination implementation for the requested provider.""" + raise NotImplementedError diff --git a/litellm/integrations/focus/destinations/s3_destination.py b/litellm/integrations/focus/destinations/s3_destination.py new file mode 100644 index 00000000000..7ead87baa29 --- /dev/null +++ b/litellm/integrations/focus/destinations/s3_destination.py @@ -0,0 +1,32 @@ +"""S3 destination implementation for Focus export.""" + +from __future__ import annotations + +from typing import Any, Optional + +from .base import FocusDestination, FocusTimeWindow + + +class FocusS3Destination(FocusDestination): + """Handles uploading serialized exports to S3 buckets.""" + + def __init__( + self, + *, + prefix: str, + config: Optional[dict[str, Any]] = None, + ) -> None: + self.prefix = prefix.rstrip("/") + self.config = config or {} + + async def deliver( + self, + *, + content: bytes, + time_window: FocusTimeWindow, + filename: str, + ) -> None: + raise NotImplementedError + + def _build_object_key(self, *, time_window: FocusTimeWindow, filename: str) -> str: + raise NotImplementedError diff --git a/litellm/integrations/focus/focus_export_logger.py b/litellm/integrations/focus/focus_export_logger.py new file mode 100644 index 00000000000..723dbc706fb --- /dev/null +++ b/litellm/integrations/focus/focus_export_logger.py @@ -0,0 +1,129 @@ +"""Focus export logger orchestrating DB pull/transform/upload.""" + +from __future__ import annotations + +import os +from datetime import datetime, timedelta, timezone +from typing import TYPE_CHECKING, Any, Optional + +import polars as pl + +from litellm.integrations.custom_logger import CustomLogger + +from .destinations import ( + FocusDestination, + FocusDestinationFactory, + FocusTimeWindow, +) +from .serializers import FocusParquetSerializer, FocusSerializer +from .transformer import FocusTransformer + +if TYPE_CHECKING: + from apscheduler.schedulers.asyncio import AsyncIOScheduler +else: + AsyncIOScheduler = Any + + +class FocusExportLogger(CustomLogger): + """Coordinates Focus export jobs across transformer/serializer/destination layers.""" + + def __init__( + self, + *, + provider: Optional[str] = None, + export_format: Optional[str] = None, + frequency: Optional[str] = None, + cron_offset_minute: Optional[int] = None, + interval_seconds: Optional[int] = None, + prefix: Optional[str] = None, + destination_config: Optional[dict[str, Any]] = None, + **kwargs: Any, + ) -> None: + super().__init__(**kwargs) + self.provider = (provider or os.getenv("FOCUS_EXPORT_PROVIDER") or "s3").lower() + self.export_format = ( + export_format or os.getenv("FOCUS_EXPORT_FORMAT") or "parquet" + ).lower() + self.frequency = ( + frequency or os.getenv("FOCUS_EXPORT_FREQUENCY") or "hourly" + ).lower() + self.cron_offset_minute = ( + cron_offset_minute + if cron_offset_minute is not None + else int(os.getenv("FOCUS_EXPORT_CRON_OFFSET", "5")) + ) + self.interval_seconds = ( + interval_seconds + if interval_seconds is not None + else os.getenv("FOCUS_EXPORT_INTERVAL_SECONDS") + ) + self.prefix = prefix or os.getenv("FOCUS_EXPORT_PREFIX", "focus_exports") + + self._destination = self._init_destination( + destination_config=destination_config, + ) + self._serializer = self._init_serializer() + self._transformer = FocusTransformer() + + def _init_serializer(self) -> FocusSerializer: + """Return serializer implementation for requested format.""" + if self.export_format != "parquet": + raise NotImplementedError("Only parquet export supported currently") + return FocusParquetSerializer() + + def _init_destination( + self, + *, + destination_config: Optional[dict[str, Any]], + ) -> FocusDestination: + """Factory for destination implementations.""" + resolved_config = self._resolve_destination_config(destination_config) + return FocusDestinationFactory.create( + provider=self.provider, + prefix=self.prefix, + config=resolved_config, + ) + + def _resolve_destination_config( + self, + destination_config: Optional[dict[str, Any]], + ) -> dict[str, Any]: + """Collect provider-specific configuration for destination creation.""" + raise NotImplementedError + + async def export_usage_data(self) -> None: + """Public hook to trigger export immediately.""" + raise NotImplementedError + + async def dry_run_export_usage_data(self) -> dict: + """Return transformed data without uploading.""" + raise NotImplementedError + + async def initialize_focus_export_job(self) -> None: + """Entry point for scheduler jobs to run export cycle with locking.""" + raise NotImplementedError + + @staticmethod + async def init_focus_export_background_job( + scheduler: AsyncIOScheduler, + ) -> None: + """Register the export cron/interval job with the provided scheduler.""" + raise NotImplementedError + + def _compute_time_window(self, now: datetime) -> FocusTimeWindow: + """Derive the time window to export based on configured frequency.""" + raise NotImplementedError + + def _serialize_and_upload( + self, + frame: pl.DataFrame, + window: FocusTimeWindow, + ) -> None: + """Helper stub for serializing and delegating to destination.""" + raise NotImplementedError + + def _build_filename(self) -> str: + """Return the canonical file name for exports.""" + if not self._serializer.extension: + raise ValueError("Serializer must declare a file extension") + return f"usage.{self._serializer.extension}" diff --git a/litellm/integrations/focus/serializers/__init__.py b/litellm/integrations/focus/serializers/__init__.py new file mode 100644 index 00000000000..18187bf73e5 --- /dev/null +++ b/litellm/integrations/focus/serializers/__init__.py @@ -0,0 +1,6 @@ +"""Serializer package exports for Focus integration.""" + +from .base import FocusSerializer +from .parquet import FocusParquetSerializer + +__all__ = ["FocusSerializer", "FocusParquetSerializer"] diff --git a/litellm/integrations/focus/serializers/base.py b/litellm/integrations/focus/serializers/base.py new file mode 100644 index 00000000000..6da080dae81 --- /dev/null +++ b/litellm/integrations/focus/serializers/base.py @@ -0,0 +1,18 @@ +"""Serializer abstractions for Focus export.""" + +from __future__ import annotations + +from abc import ABC, abstractmethod + +import polars as pl + + +class FocusSerializer(ABC): + """Base serializer turning Focus frames into bytes.""" + + extension: str = "" + + @abstractmethod + def serialize(self, frame: pl.DataFrame) -> bytes: + """Convert the normalized Focus frame into the chosen format.""" + raise NotImplementedError diff --git a/litellm/integrations/focus/serializers/parquet.py b/litellm/integrations/focus/serializers/parquet.py new file mode 100644 index 00000000000..3d42337e388 --- /dev/null +++ b/litellm/integrations/focus/serializers/parquet.py @@ -0,0 +1,16 @@ +"""Parquet serializer for Focus export.""" + +from __future__ import annotations + +import polars as pl + +from .base import FocusSerializer + + +class FocusParquetSerializer(FocusSerializer): + """Placeholder Parquet serializer implementation.""" + + extension = "parquet" + + def serialize(self, frame: pl.DataFrame) -> bytes: + raise NotImplementedError diff --git a/litellm/integrations/focus/transformer.py b/litellm/integrations/focus/transformer.py new file mode 100644 index 00000000000..45a77dd8ac5 --- /dev/null +++ b/litellm/integrations/focus/transformer.py @@ -0,0 +1,13 @@ +"""Focus export data transformer.""" + +from __future__ import annotations + +import polars as pl + + +class FocusTransformer: + """Transforms LiteLLM DB rows into Focus-compatible schema.""" + + def transform(self, frame: pl.DataFrame) -> pl.DataFrame: + """Return a normalized frame expected by downstream serializers.""" + raise NotImplementedError