feat(focus): add GCS destination for FOCUS export (#29751)

* test: add failing tests for FocusGCSDestination

* feat: add FocusGCSDestination reusing GCSBucketBase auth

* feat: register FocusGCSDestination in factory; export from __init__

* fix(focus): preserve GCS_PATH_SERVICE_ACCOUNT when service_account_json not in config

* style: apply Black formatting to gcs_destination and tests

* style: apply Black formatting to factory.py

litellm/integrations/focus/destinations/__init__.py

@@ -2,12 +2,14 @@
 
 from .base import FocusDestination, FocusTimeWindow
 from .factory import FocusDestinationFactory
+from .gcs_destination import FocusGCSDestination
 from .s3_destination import FocusS3Destination
 from .vantage_destination import FocusVantageDestination
 
 __all__ = [
     "FocusDestination",
     "FocusDestinationFactory",
+    "FocusGCSDestination",
     "FocusTimeWindow",
     "FocusS3Destination",
     "FocusVantageDestination",

litellm/integrations/focus/destinations/factory.py

@@ -6,6 +6,7 @@ import os
 from typing import Any, Dict, Optional
 
 from .base import FocusDestination
+from .gcs_destination import FocusGCSDestination
 from .s3_destination import FocusS3Destination
 from .vantage_destination import FocusVantageDestination
 
@@ -29,6 +30,8 @@ class FocusDestinationFactory:
             return FocusS3Destination(prefix=prefix, config=normalized_config)
         if provider_lower == "vantage":
             return FocusVantageDestination(prefix=prefix, config=normalized_config)
+        if provider_lower == "gcs":
+            return FocusGCSDestination(prefix=prefix, config=normalized_config)
         raise NotImplementedError(
             f"Provider '{provider}' not supported for Focus export"
         )
@@ -72,6 +75,18 @@ class FocusDestinationFactory:
                     "VANTAGE_INTEGRATION_TOKEN must be provided for Vantage exports"
                 )
             return {k: v for k, v in resolved.items() if v is not None}
+        if provider == "gcs":
+            resolved = {
+                "bucket_name": overrides.get("bucket_name")
+                or os.getenv("FOCUS_GCS_BUCKET_NAME"),
+                "service_account_json": overrides.get("service_account_json")
+                or os.getenv("FOCUS_GCS_PATH_SERVICE_ACCOUNT"),
+            }
+            if not resolved.get("bucket_name"):
+                raise ValueError(
+                    "FOCUS_GCS_BUCKET_NAME must be provided for GCS exports"
+                )
+            return {k: v for k, v in resolved.items() if v is not None}
         raise NotImplementedError(
             f"Provider '{provider}' not supported for Focus export configuration"
         )

litellm/integrations/focus/destinations/gcs_destination.py

@@ -0,0 +1,74 @@
+"""GCS destination for Focus export — reuses GCSBucketBase auth and httpx client."""
+
+from __future__ import annotations
+
+from datetime import timezone
+from typing import Any, Optional
+
+from litellm._logging import verbose_logger
+from litellm.integrations.gcs_bucket.gcs_bucket_base import GCSBucketBase
+from litellm.litellm_core_utils.cloud_storage_security import (
+    encode_gcs_object_name_for_url,
+)
+
+from .base import FocusDestination, FocusTimeWindow
+
+
+class FocusGCSDestination(GCSBucketBase, FocusDestination):
+    """Upload serialized Focus exports to GCS using the GCS JSON API."""
+
+    def __init__(
+        self,
+        *,
+        prefix: str,
+        config: Optional[dict[str, Any]] = None,
+    ) -> None:
+        config = config or {}
+        bucket_name = config.get("bucket_name")
+        if not bucket_name:
+            raise ValueError("bucket_name must be provided for GCS destination")
+        super().__init__(bucket_name=bucket_name)
+        service_account_json = config.get("service_account_json")
+        if service_account_json is not None:
+            self.path_service_account_json = service_account_json
+        self.prefix = prefix.rstrip("/")
+
+    async def deliver(
+        self,
+        *,
+        content: bytes,
+        time_window: FocusTimeWindow,
+        filename: str,
+    ) -> None:
+        object_name = self._build_object_key(time_window=time_window, filename=filename)
+        headers = await self.construct_request_headers(
+            service_account_json=self.path_service_account_json
+        )
+        headers["Content-Type"] = "application/octet-stream"
+        encoded_name = encode_gcs_object_name_for_url(object_name)
+        url = (
+            f"https://storage.googleapis.com/upload/storage/v1/b/"
+            f"{self.BUCKET_NAME}/o?uploadType=media&name={encoded_name}"
+        )
+        response = await self.async_httpx_client.post(
+            url=url, headers=headers, data=content
+        )
+        if response.status_code != 200:
+            raise RuntimeError(
+                f"GCS upload failed: status={response.status_code} body={response.text}"
+            )
+        verbose_logger.debug(
+            "Focus GCS: uploaded %d bytes to gs://%s/%s",
+            len(content),
+            self.BUCKET_NAME,
+            object_name,
+        )
+
+    def _build_object_key(self, *, time_window: FocusTimeWindow, filename: str) -> str:
+        start_utc = time_window.start_time.astimezone(timezone.utc)
+        date_component = f"date={start_utc.strftime('%Y-%m-%d')}"
+        parts = [self.prefix, date_component]
+        if time_window.frequency == "hourly":
+            parts.append(f"hour={start_utc.strftime('%H')}")
+        key_prefix = "/".join(filter(None, parts))
+        return f"{key_prefix}/{filename}" if key_prefix else filename

tests/test_litellm/integrations/focus/test_focus_gcs_destination.py

@@ -0,0 +1,180 @@
+"""Tests for FocusGCSDestination."""
+
+from __future__ import annotations
+
+from datetime import datetime, timezone
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+
+from litellm.integrations.focus.destinations.base import FocusTimeWindow
+
+
+def _make_window(frequency: str = "hourly") -> FocusTimeWindow:
+    return FocusTimeWindow(
+        start_time=datetime(2026, 1, 1, 10, 0, 0, tzinfo=timezone.utc),
+        end_time=datetime(2026, 1, 1, 11, 0, 0, tzinfo=timezone.utc),
+        frequency=frequency,
+    )
+
+
+@pytest.mark.asyncio
+async def test_deliver_posts_to_gcs_upload_endpoint():
+    """deliver() must POST raw bytes to the GCS upload endpoint."""
+    from litellm.integrations.focus.destinations.gcs_destination import (
+        FocusGCSDestination,
+    )
+
+    dest = FocusGCSDestination(
+        prefix="focus_exports",
+        config={"bucket_name": "my-bucket", "service_account_json": None},
+    )
+
+    mock_response = MagicMock()
+    mock_response.status_code = 200
+
+    mock_client = MagicMock()
+    mock_client.post = AsyncMock(return_value=mock_response)
+    dest.async_httpx_client = mock_client
+
+    with patch.object(
+        dest,
+        "construct_request_headers",
+        new=AsyncMock(return_value={"Authorization": "Bearer tok-123"}),
+    ):
+        await dest.deliver(
+            content=b"col1,col2\nval1,val2\n",
+            time_window=_make_window(),
+            filename="usage_20260101T100000Z_20260101T110000Z.csv",
+        )
+
+    mock_client.post.assert_called_once()
+    call_kwargs = mock_client.post.call_args
+    url = call_kwargs.kwargs.get("url") or call_kwargs.args[0]
+    assert "my-bucket" in url
+    assert "uploadType=media" in url
+    headers = call_kwargs.kwargs["headers"]
+    assert headers["Authorization"] == "Bearer tok-123"
+
+
+@pytest.mark.asyncio
+async def test_deliver_raises_on_gcs_error():
+    """deliver() must raise RuntimeError when GCS returns non-200."""
+    from litellm.integrations.focus.destinations.gcs_destination import (
+        FocusGCSDestination,
+    )
+
+    dest = FocusGCSDestination(
+        prefix="focus_exports",
+        config={"bucket_name": "my-bucket"},
+    )
+
+    mock_response = MagicMock()
+    mock_response.status_code = 403
+    mock_response.text = "Permission denied"
+
+    mock_client = MagicMock()
+    mock_client.post = AsyncMock(return_value=mock_response)
+    dest.async_httpx_client = mock_client
+
+    with patch.object(
+        dest,
+        "construct_request_headers",
+        new=AsyncMock(return_value={"Authorization": "Bearer tok-bad"}),
+    ):
+        with pytest.raises(RuntimeError, match="GCS upload failed"):
+            await dest.deliver(
+                content=b"data",
+                time_window=_make_window(),
+                filename="usage.csv",
+            )
+
+
+def test_build_object_key_hourly():
+    """Hourly key must include date= and hour= components."""
+    from litellm.integrations.focus.destinations.gcs_destination import (
+        FocusGCSDestination,
+    )
+
+    dest = FocusGCSDestination(prefix="focus_exports", config={"bucket_name": "b"})
+    key = dest._build_object_key(
+        time_window=_make_window("hourly"), filename="usage.parquet"
+    )
+
+    assert key == "focus_exports/date=2026-01-01/hour=10/usage.parquet"
+
+
+def test_build_object_key_daily():
+    """Daily key must include date= but not hour=."""
+    from litellm.integrations.focus.destinations.gcs_destination import (
+        FocusGCSDestination,
+    )
+
+    dest = FocusGCSDestination(prefix="focus_exports", config={"bucket_name": "b"})
+    window = FocusTimeWindow(
+        start_time=datetime(2026, 1, 1, 0, 0, 0, tzinfo=timezone.utc),
+        end_time=datetime(2026, 1, 2, 0, 0, 0, tzinfo=timezone.utc),
+        frequency="daily",
+    )
+    key = dest._build_object_key(time_window=window, filename="usage.parquet")
+
+    assert key == "focus_exports/date=2026-01-01/usage.parquet"
+
+
+def test_missing_bucket_name_raises():
+    """Constructing without bucket_name must raise ValueError."""
+    from litellm.integrations.focus.destinations.gcs_destination import (
+        FocusGCSDestination,
+    )
+
+    with pytest.raises(ValueError, match="bucket_name"):
+        FocusGCSDestination(prefix="focus_exports", config={})
+
+
+def test_global_gcs_service_account_not_overwritten_when_absent(monkeypatch):
+    """service_account_json absent from config must not overwrite GCS_PATH_SERVICE_ACCOUNT.
+
+    GCSBucketBase sets self.path_service_account_json from GCS_PATH_SERVICE_ACCOUNT.
+    If config has no service_account_json key, we must leave the parent value intact
+    so deployments using the global credential don't silently fall back to ADC.
+    """
+    monkeypatch.setenv("GCS_PATH_SERVICE_ACCOUNT", "/global/sa.json")
+
+    from litellm.integrations.focus.destinations.gcs_destination import (
+        FocusGCSDestination,
+    )
+
+    dest = FocusGCSDestination(prefix="focus_exports", config={"bucket_name": "b"})
+
+    assert dest.path_service_account_json == "/global/sa.json"
+
+
+def test_explicit_service_account_overrides_global(monkeypatch):
+    """Explicit service_account_json in config must take precedence over GCS_PATH_SERVICE_ACCOUNT."""
+    monkeypatch.setenv("GCS_PATH_SERVICE_ACCOUNT", "/global/sa.json")
+
+    from litellm.integrations.focus.destinations.gcs_destination import (
+        FocusGCSDestination,
+    )
+
+    dest = FocusGCSDestination(
+        prefix="focus_exports",
+        config={"bucket_name": "b", "service_account_json": "/focus/sa.json"},
+    )
+
+    assert dest.path_service_account_json == "/focus/sa.json"
+
+
+def test_factory_creates_gcs_destination(monkeypatch):
+    """FocusDestinationFactory.create(provider='gcs') must return FocusGCSDestination."""
+    monkeypatch.setenv("FOCUS_GCS_BUCKET_NAME", "env-bucket")
+
+    from litellm.integrations.focus.destinations.factory import FocusDestinationFactory
+    from litellm.integrations.focus.destinations.gcs_destination import (
+        FocusGCSDestination,
+    )
+
+    dest = FocusDestinationFactory.create(provider="gcs", prefix="focus_exports")
+
+    assert isinstance(dest, FocusGCSDestination)
+    assert dest.BUCKET_NAME == "env-bucket"