first

app/routers/ga_admin.py

@@ -0,0 +1,104 @@
+"""Google Analytics 4 - Admin API (read: accounts, properties, data streams).
+
+https://developers.google.com/analytics/devguides/config/admin/v1
+
+Credentials: X-GA-Access-Token (preferred) or X-GA-Credentials.
+"""
+from __future__ import annotations
+
+from typing import Any
+
+from fastapi import APIRouter, Depends, Path, Query
+
+from ..clients.ga_client import GoogleAnalyticsClient
+from ..credentials import GaCredentials, get_ga_credentials
+
+router = APIRouter(prefix="/ga/admin", tags=["google-analytics: admin"])
+
+
+def _property(property_id: str) -> str:
+    pid = property_id.strip()
+    return pid if pid.startswith("properties/") else f"properties/{pid}"
+
+
+@router.get("/accounts", summary="List accessible GA4 accounts")
+async def list_accounts(
+    page_size: int | None = Query(None, ge=1, le=200, alias="pageSize"),
+    page_token: str | None = Query(None, alias="pageToken"),
+    creds: GaCredentials = Depends(get_ga_credentials),
+) -> Any:
+    params = _paging(page_size, page_token)
+    client = GoogleAnalyticsClient(creds)
+    return await client.admin_get("/accounts", params=params)
+
+
+@router.get(
+    "/accountSummaries",
+    summary="List account summaries (accounts + their properties)",
+)
+async def list_account_summaries(
+    page_size: int | None = Query(None, ge=1, le=200, alias="pageSize"),
+    page_token: str | None = Query(None, alias="pageToken"),
+    creds: GaCredentials = Depends(get_ga_credentials),
+) -> Any:
+    params = _paging(page_size, page_token)
+    client = GoogleAnalyticsClient(creds)
+    return await client.admin_get("/accountSummaries", params=params)
+
+
+@router.get("/properties", summary="List properties under an account")
+async def list_properties(
+    account_id: str = Query(
+        ...,
+        alias="accountId",
+        description="Numeric account id; the filter parent:accounts/{id} is built for you.",
+    ),
+    page_size: int | None = Query(None, ge=1, le=200, alias="pageSize"),
+    page_token: str | None = Query(None, alias="pageToken"),
+    show_deleted: bool | None = Query(None, alias="showDeleted"),
+    creds: GaCredentials = Depends(get_ga_credentials),
+) -> Any:
+    params: dict[str, Any] = {"filter": f"parent:accounts/{account_id.strip()}"}
+    params.update(_paging(page_size, page_token))
+    if show_deleted is not None:
+        params["showDeleted"] = show_deleted
+    client = GoogleAnalyticsClient(creds)
+    return await client.admin_get("/properties", params=params)
+
+
+@router.get(
+    "/properties/{property_id}",
+    summary="Get a single GA4 property",
+)
+async def get_property(
+    property_id: str = Path(..., description="GA4 property id"),
+    creds: GaCredentials = Depends(get_ga_credentials),
+) -> Any:
+    client = GoogleAnalyticsClient(creds)
+    return await client.admin_get(f"/{_property(property_id)}")
+
+
+@router.get(
+    "/properties/{property_id}/dataStreams",
+    summary="List data streams of a GA4 property",
+)
+async def list_data_streams(
+    property_id: str = Path(..., description="GA4 property id"),
+    page_size: int | None = Query(None, ge=1, le=200, alias="pageSize"),
+    page_token: str | None = Query(None, alias="pageToken"),
+    creds: GaCredentials = Depends(get_ga_credentials),
+) -> Any:
+    params = _paging(page_size, page_token)
+    client = GoogleAnalyticsClient(creds)
+    return await client.admin_get(
+        f"/{_property(property_id)}/dataStreams", params=params
+    )
+
+
+def _paging(page_size: int | None, page_token: str | None) -> dict[str, Any]:
+    params: dict[str, Any] = {}
+    if page_size is not None:
+        params["pageSize"] = page_size
+    if page_token:
+        params["pageToken"] = page_token
+    return params