JiriUhlir
105 lines
3.6 KiB
Python
105 lines
3.6 KiB
Python
"""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
|