diff --git a/README.md b/README.md index cd91486..14fbda1 100644 --- a/README.md +++ b/README.md @@ -1,3 +1,60 @@ # analytics -Generated by AppFactory. +Stateless API proxy for **Google Analytics 4** and **Sklik** (Seznam), running +in AppFactory behind the Caddy reverse proxy at `/apps/analytics`. + +The service stores no secrets. Every credential is supplied **per request** as +an `X-` header and used only to call the upstream API. + +## Endpoints + +Interactive docs (Swagger UI): `/docs` — publicly `https://services.csbot.cz/apps/analytics/docs`. + +| Area | Method | Path | +| --- | --- | --- | +| Meta | GET | `/health`, `/version` | +| GA4 Data | POST | `/ga/data/properties/{id}/runReport` | +| GA4 Data | POST | `/ga/data/properties/{id}/runPivotReport` | +| GA4 Data | POST | `/ga/data/properties/{id}/batchRunReports` | +| GA4 Data | POST | `/ga/data/properties/{id}/batchRunPivotReports` | +| GA4 Data | POST | `/ga/data/properties/{id}/runRealtimeReport` | +| GA4 Data | POST | `/ga/data/properties/{id}/checkCompatibility` | +| GA4 Data | GET | `/ga/data/properties/{id}/metadata` | +| GA4 Admin | GET | `/ga/admin/accounts`, `/ga/admin/accountSummaries` | +| GA4 Admin | GET | `/ga/admin/properties` (`?accountId=`), `/ga/admin/properties/{id}` | +| GA4 Admin | GET | `/ga/admin/properties/{id}/dataStreams` | +| Sklik | POST | `/sklik/login`, `/sklik/report/{entity}`, `/sklik/rpc/{method}` | +| Sklik | GET | `/sklik/limits` | + +## Credentials (headers) + +**Google Analytics** — token wins over service account: + +| Header | Required | Meaning | +| --- | --- | --- | +| `X-GA-Access-Token` | one of these | Ready OAuth2 access token (used as Bearer). | +| `X-GA-Credentials` | one of these | Base64-encoded service-account JSON key; the proxy mints a token. | +| `X-GA-Quota-Project` | no | Google Cloud project id for quota/billing. | + +**Sklik:** + +| Header | Required | Meaning | +| --- | --- | --- | +| `X-Sklik-Token` | yes | Sklik API token from account settings. | +| `X-Sklik-User-Id` | no | Managed account id for agency/MCC access. | + +## Run locally + +```bash +pip install -r requirements.txt +uvicorn app.main:app --reload --port 8000 +# open http://localhost:8000/docs +``` + +## Configuration (env) + +Non-secret only — see [app/config.py](app/config.py): `ROOT_PATH`, +`GA_DATA_BASE_URL`, `GA_ADMIN_BASE_URL`, `GA_SCOPE`, `SKLIK_BASE_URL`, +`HTTP_TIMEOUT_SECONDS`, `LOG_LEVEL`. + +See [documentation/](documentation/) for per-integration detail. diff --git a/app/clients/__init__.py b/app/clients/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/app/clients/ga_client.py b/app/clients/ga_client.py new file mode 100644 index 0000000..a02870c --- /dev/null +++ b/app/clients/ga_client.py @@ -0,0 +1,180 @@ +"""Google Analytics client. + +Thin proxy over the GA4 Data API and Admin API. Request/response bodies are +forwarded as-is so callers keep the full flexibility of Google's API; this +module only handles authentication (Bearer token), the base URL, and error +mapping. + +Authentication (per chosen model, token wins over service account): + * If ``X-GA-Access-Token`` was supplied, it is used directly. + * Otherwise a short-lived access token is minted from the service-account + JSON via google-auth and cached in-memory (keyed by key id + scope) until + shortly before it expires. The key material is never written to disk or log. +""" +from __future__ import annotations + +import hashlib +import threading +import time +from typing import Any + +import httpx +from fastapi.concurrency import run_in_threadpool + +from .. import config +from ..credentials import GaCredentials +from ..errors import MissingCredentialsError, UpstreamError +from ..logging_config import get_logger + +logger = get_logger(__name__) + +# In-memory access-token cache: { cache_key: (token, expiry_epoch_seconds) }. +# Memory only - mirrors the stateless design (no secret ever persisted). +_token_cache: dict[str, tuple[str, float]] = {} +_token_lock = threading.Lock() + +# Refresh a minted token this many seconds before its real expiry. +_EXPIRY_SKEW = 60.0 + + +def _mint_token_sync(info: dict, scope: str) -> tuple[str, float]: + """Mint an OAuth2 access token from a service-account key (blocking).""" + # Imported lazily so the module imports even if google-auth is missing, + # and so the dependency is only needed when service-account auth is used. + from google.auth.transport.requests import Request + from google.oauth2 import service_account + + try: + creds = service_account.Credentials.from_service_account_info( + info, scopes=[scope] + ) + except (ValueError, KeyError) as exc: + raise MissingCredentialsError( + f"X-GA-Credentials is not a usable service-account key: {exc}" + ) from exc + + try: + creds.refresh(Request()) + except Exception as exc: # google.auth.exceptions.RefreshError and friends + # Surface as upstream auth failure - do NOT log the key material. + raise UpstreamError( + f"Failed to obtain Google access token from service account: {exc}", + status=401, + ) from exc + + expiry = creds.expiry.timestamp() if creds.expiry else (time.time() + 3600) + return creds.token, expiry + + +def _cache_key(info: dict, scope: str) -> str: + # Identify a key by its private_key_id + client_email + scope. Hashed so the + # raw identifiers never sit in a dict key we might later log. + raw = f"{info.get('private_key_id', '')}|{info.get('client_email', '')}|{scope}" + return hashlib.sha256(raw.encode("utf-8")).hexdigest() + + +async def _bearer_token(creds: GaCredentials) -> str: + if creds.access_token: + return creds.access_token + + if creds.service_account_info is None: + # get_ga_credentials guarantees one of the two, but be defensive. + raise MissingCredentialsError( + "No GA access token and no service-account credentials available." + ) + + scope = config.GA_SCOPE + key = _cache_key(creds.service_account_info, scope) + now = time.time() + + with _token_lock: + cached = _token_cache.get(key) + if cached and cached[1] - _EXPIRY_SKEW > now: + return cached[0] + + # Mint outside the lock (network call); google-auth is blocking, so offload + # it to a thread to avoid stalling the event loop. + token, expiry = await run_in_threadpool( + _mint_token_sync, creds.service_account_info, scope + ) + with _token_lock: + _token_cache[key] = (token, expiry) + return token + + +class GoogleAnalyticsClient: + """Authenticated HTTP client for the GA4 Data and Admin APIs.""" + + def __init__(self, creds: GaCredentials) -> None: + self._creds = creds + + async def _request( + self, + method: str, + base_url: str, + path: str, + *, + params: dict | None = None, + json_body: Any | None = None, + ) -> Any: + token = await _bearer_token(self._creds) + headers = {"Authorization": f"Bearer {token}"} + if self._creds.quota_project: + headers["x-goog-user-project"] = self._creds.quota_project + + url = f"{base_url}{path}" + try: + async with httpx.AsyncClient( + timeout=config.HTTP_TIMEOUT_SECONDS + ) as client: + resp = await client.request( + method, url, params=params, json=json_body, headers=headers + ) + except httpx.TimeoutException as exc: + raise UpstreamError( + "Google Analytics request timed out.", status=504 + ) from exc + except httpx.HTTPError as exc: + raise UpstreamError( + f"Google Analytics is unreachable: {exc}", status=502 + ) from exc + + return _parse_google_response(resp) + + # --- Data API ------------------------------------------------------------- + async def data_post(self, path: str, body: Any) -> Any: + return await self._request( + "POST", config.GA_DATA_BASE_URL, path, json_body=body + ) + + async def data_get(self, path: str, params: dict | None = None) -> Any: + return await self._request( + "GET", config.GA_DATA_BASE_URL, path, params=params + ) + + # --- Admin API ------------------------------------------------------------ + async def admin_get(self, path: str, params: dict | None = None) -> Any: + return await self._request( + "GET", config.GA_ADMIN_BASE_URL, path, params=params + ) + + +def _parse_google_response(resp: httpx.Response) -> Any: + try: + payload = resp.json() + except ValueError: + payload = {"raw": resp.text} + + if resp.is_success: + return payload + + # Google returns {"error": {"code", "message", "status", ...}}. + message = "Google Analytics API error" + if isinstance(payload, dict) and isinstance(payload.get("error"), dict): + message = payload["error"].get("message", message) + raise UpstreamError( + message, + status=502 if resp.status_code >= 500 else resp.status_code, + upstream_status=resp.status_code, + body=payload, + ) diff --git a/app/clients/sklik_client.py b/app/clients/sklik_client.py new file mode 100644 index 0000000..cf945ae --- /dev/null +++ b/app/clients/sklik_client.py @@ -0,0 +1,193 @@ +"""Sklik (Seznam) "Drak" JSON API client. + +Protocol (verified against the official seznam/api-examples JSON example): + * Endpoint: ``{SKLIK_BASE_URL}/{method}`` e.g. .../drak/json/v5/campaigns.list + * HTTP POST, body = a JSON ARRAY of positional arguments. + * ``client.loginByToken`` takes the API token as its single argument and + returns ``{"status":200,"session":"...",...}``. + * Every authenticated method takes the user struct ``{"session": ...}`` + (optionally ``"userId"``) as its FIRST argument, followed by the method's + own arguments. + * Every response is an object containing ``status`` (HTTP-style int), + ``statusMessage``, a refreshed ``session``, plus method-specific data. + +The proxy is stateless: it logs in with ``X-Sklik-Token`` per request to obtain +a session, then performs the requested call. The token and session are never +logged. +""" +from __future__ import annotations + +from typing import Any + +import httpx + +from .. import config +from ..errors import UpstreamError +from ..logging_config import get_logger + +logger = get_logger(__name__) + +# Sklik report data is paginated; readReport is called with an offset/limit +# window until all rows are fetched. Keep the page size conservative. +_REPORT_PAGE_LIMIT = 100 +# Hard stop so a misbehaving upstream can't loop forever. +_REPORT_MAX_PAGES = 1000 + + +class SklikClient: + """Performs JSON-RPC calls against the Sklik Drak API.""" + + def __init__(self, token: str, user_id: int | None = None) -> None: + self._token = token + self._user_id = user_id + self._session: str | None = None + + async def __aenter__(self) -> "SklikClient": + self._http = httpx.AsyncClient(timeout=config.HTTP_TIMEOUT_SECONDS) + return self + + async def __aexit__(self, *exc: Any) -> None: + await self._http.aclose() + + async def _call(self, method: str, args: list[Any]) -> dict: + """Low-level: POST a JSON array of args to ``/{method}``.""" + url = f"{config.SKLIK_BASE_URL}/{method}" + try: + resp = await self._http.post(url, json=args) + except httpx.TimeoutException as exc: + raise UpstreamError( + f"Sklik request timed out ({method}).", status=504 + ) from exc + except httpx.HTTPError as exc: + raise UpstreamError( + f"Sklik is unreachable ({method}): {exc}", status=502 + ) from exc + + try: + payload = resp.json() + except ValueError as exc: + raise UpstreamError( + f"Sklik returned a non-JSON response ({method}).", + status=502, + upstream_status=resp.status_code, + body={"raw": resp.text}, + ) from exc + + if not isinstance(payload, dict): + raise UpstreamError( + f"Unexpected Sklik response shape ({method}).", + status=502, + body=payload, + ) + + status = payload.get("status") + # Sklik conveys business errors in the body with an HTTP-style status. + # 200 OK, 206 partially OK, 301 "user is serviced" are all acceptable. + if status not in (200, 206, 301): + raise UpstreamError( + payload.get("statusMessage", f"Sklik error on {method}."), + status=400 if isinstance(status, int) and 400 <= status < 500 else 502, + upstream_status=status if isinstance(status, int) else None, + body=payload, + ) + + # Refresh our session from every response (Sklik rotates it). + new_session = payload.get("session") + if isinstance(new_session, str) and new_session: + self._session = new_session + return payload + + async def login(self) -> dict: + """Exchange the API token for a session. Idempotent per client.""" + payload = await self._call("client.loginByToken", [self._token]) + if not self._session: + raise UpstreamError( + "Sklik login succeeded but returned no session.", + status=502, + body=payload, + ) + return payload + + def _user_struct(self) -> dict: + user: dict[str, Any] = {"session": self._session} + if self._user_id is not None: + user["userId"] = self._user_id + return user + + async def call(self, method: str, args: list[Any] | None = None) -> dict: + """Authenticated call: prepends the user/session struct to ``args``. + + Logs in first if no session is held yet. ``method`` is e.g. + ``campaigns.list``; ``args`` are the method arguments AFTER the user + struct. + """ + if method == "client.loginByToken": + # Login is handled by login(); never forward the bare token here. + raise UpstreamError( + "client.loginByToken cannot be called directly; the proxy " + "manages the session.", + status=400, + ) + if not self._session: + await self.login() + full_args = [self._user_struct()] + list(args or []) + return await self._call(method, full_args) + + async def fetch_report( + self, entity: str, report_args: list[Any] + ) -> dict: + """Create a stats report for ``entity`` then read all of its rows. + + ``entity`` is e.g. ``campaigns``/``groups``/``ads``/``keywords``. + Calls ``{entity}.createReport`` with ``report_args`` (the restriction + + display-options structs), then pages through ``{entity}.readReport`` + until every row is collected. + """ + created = await self.call(f"{entity}.createReport", report_args) + report_id = created.get("reportId") + if not report_id: + raise UpstreamError( + f"{entity}.createReport returned no reportId.", + status=502, + body=created, + ) + total = created.get("totalCount", 0) + + rows: list[Any] = [] + offset = 0 + pages = 0 + while True: + page = await self.call( + f"{entity}.readReport", + [ + report_id, + { + "offset": offset, + "limit": _REPORT_PAGE_LIMIT, + "allowEmptyStatistics": False, + }, + ], + ) + batch = page.get("report") or [] + rows.extend(batch) + pages += 1 + offset += _REPORT_PAGE_LIMIT + if len(batch) < _REPORT_PAGE_LIMIT: + break + if pages >= _REPORT_MAX_PAGES: + logger.warning( + "Sklik %s.readReport hit the %d-page safety cap (collected " + "%d rows); result may be truncated.", + entity, + _REPORT_MAX_PAGES, + len(rows), + ) + break + + return { + "reportId": report_id, + "totalCount": total, + "returnedCount": len(rows), + "truncated": pages >= _REPORT_MAX_PAGES, + "report": rows, + } diff --git a/app/config.py b/app/config.py new file mode 100644 index 0000000..9306d96 --- /dev/null +++ b/app/config.py @@ -0,0 +1,41 @@ +"""Runtime configuration read from environment variables. + +AppFactory injects variables/secrets as environment variables (see AGENTS.md). +This module holds only NON-secret infrastructure configuration. Per-request +credentials are never stored here - they arrive in X- headers (see +``app.credentials``). +""" +import os + +# Public app metadata +APP_NAME = os.getenv("APP_NAME", "analytics") +APP_VERSION = os.getenv("APP_VERSION", "1.0.0") + +# Reverse-proxy prefix injected by AppFactory (e.g. "/apps/analytics"). +# Empty when running locally at the domain root. +ROOT_PATH = os.getenv("ROOT_PATH", "") + +# --- Google Analytics --------------------------------------------------------- +# Base URLs are configurable so we can point at a staging/mock endpoint, but +# default to the production Google endpoints. +GA_DATA_BASE_URL = os.getenv( + "GA_DATA_BASE_URL", "https://analyticsdata.googleapis.com/v1beta" +) +GA_ADMIN_BASE_URL = os.getenv( + "GA_ADMIN_BASE_URL", "https://analyticsadmin.googleapis.com/v1beta" +) +# OAuth scope requested when minting an access token from a service account. +# analytics.readonly is sufficient for reporting (Data API) and for listing +# accounts/properties/data streams (Admin API read operations). +GA_SCOPE = os.getenv( + "GA_SCOPE", "https://www.googleapis.com/auth/analytics.readonly" +) + +# --- Sklik (Seznam) ----------------------------------------------------------- +# Sklik "Drak" JSON API. The method name is appended to this base URL and the +# HTTP body is a JSON array of positional arguments. +SKLIK_BASE_URL = os.getenv("SKLIK_BASE_URL", "https://api.sklik.cz/drak/json/v5") + +# --- HTTP --------------------------------------------------------------------- +# Upstream request timeout in seconds. +HTTP_TIMEOUT_SECONDS = float(os.getenv("HTTP_TIMEOUT_SECONDS", "60")) diff --git a/app/credentials.py b/app/credentials.py new file mode 100644 index 0000000..1cec05b --- /dev/null +++ b/app/credentials.py @@ -0,0 +1,108 @@ +"""Per-request credential extraction from X- headers (FastAPI dependencies). + +The service is a STATELESS proxy: it stores no secrets. Every credential is +supplied per request as an X- header and used only to talk to the upstream API +(see AGENTS.md "Secrets v parametrech"). Declaring the headers as FastAPI +``Header`` parameters makes them appear per-operation in Swagger, including the +"Try it out" form. + +Google Analytics (chosen model: token has precedence over service account): + * ``X-GA-Access-Token`` - a ready OAuth2 access token; used directly as Bearer. + * ``X-GA-Credentials`` - base64-encoded service-account JSON key; the proxy + mints a short-lived access token from it. + * ``X-GA-Quota-Project`` - optional billing/quota project id. + At least one of token / credentials must be present. + +Sklik: + * ``X-Sklik-Token`` - the Sklik API token from account settings. The proxy + calls ``client.loginByToken`` to obtain a session. +""" +from __future__ import annotations + +import base64 +import binascii +import json +from dataclasses import dataclass + +from fastapi import Header + +from .errors import MissingCredentialsError + + +# --- Google Analytics --------------------------------------------------------- +@dataclass +class GaCredentials: + access_token: str | None + service_account_info: dict | None + quota_project: str | None + + +def get_ga_credentials( + x_ga_access_token: str | None = Header( + default=None, + alias="X-GA-Access-Token", + description="Ready OAuth2 access token used directly as a Bearer token. " + "Takes precedence over X-GA-Credentials.", + ), + x_ga_credentials: str | None = Header( + default=None, + alias="X-GA-Credentials", + description="Base64-encoded Google service-account JSON key. The proxy " + "mints a short-lived access token from it (scope analytics.readonly). " + "Used only if X-GA-Access-Token is absent.", + ), + x_ga_quota_project: str | None = Header( + default=None, + alias="X-GA-Quota-Project", + description="Optional Google Cloud project id used for quota/billing " + "(sets the x-goog-user-project header upstream).", + ), +) -> GaCredentials: + """Resolve GA credentials from headers. Token wins over service account.""" + access_token = (x_ga_access_token or "").strip() or None + + service_account_info: dict | None = None + raw = (x_ga_credentials or "").strip() + if raw: + try: + decoded = base64.b64decode(raw, validate=True) + except (binascii.Error, ValueError) as exc: + raise MissingCredentialsError( + "X-GA-Credentials is not valid base64." + ) from exc + try: + service_account_info = json.loads(decoded) + except (json.JSONDecodeError, UnicodeDecodeError) as exc: + raise MissingCredentialsError( + "X-GA-Credentials does not decode to valid JSON." + ) from exc + if not isinstance(service_account_info, dict): + raise MissingCredentialsError( + "X-GA-Credentials JSON must be a service-account object." + ) + + if not access_token and service_account_info is None: + raise MissingCredentialsError( + "Provide either X-GA-Access-Token or X-GA-Credentials." + ) + + return GaCredentials( + access_token=access_token, + service_account_info=service_account_info, + quota_project=(x_ga_quota_project or "").strip() or None, + ) + + +# --- Sklik -------------------------------------------------------------------- +def get_sklik_token( + x_sklik_token: str | None = Header( + default=None, + alias="X-Sklik-Token", + description="Sklik API token from Sklik account settings. The proxy uses " + "it to obtain a session via client.loginByToken.", + ), +) -> str: + token = (x_sklik_token or "").strip() + if not token: + raise MissingCredentialsError("X-Sklik-Token header is required.") + return token diff --git a/app/errors.py b/app/errors.py new file mode 100644 index 0000000..0b8c20e --- /dev/null +++ b/app/errors.py @@ -0,0 +1,83 @@ +"""Domain exceptions and FastAPI exception handlers. + +All errors are surfaced as JSON (never swallowed). Upstream failures preserve +the upstream status code and body so callers can diagnose problems. +""" +from __future__ import annotations + +from typing import Any + +from fastapi import FastAPI, Request +from fastapi.responses import JSONResponse + +from .logging_config import get_logger + +logger = get_logger(__name__) + + +class MissingCredentialsError(Exception): + """A required credential header was not supplied.""" + + def __init__(self, message: str) -> None: + super().__init__(message) + self.message = message + + +class UpstreamError(Exception): + """The upstream API (Google / Sklik) returned an error or was unreachable. + + ``status`` is the HTTP status to return to the caller. ``upstream_status`` + and ``body`` carry the upstream detail when available. + """ + + def __init__( + self, + message: str, + *, + status: int = 502, + upstream_status: int | None = None, + body: Any = None, + ) -> None: + super().__init__(message) + self.message = message + self.status = status + self.upstream_status = upstream_status + self.body = body + + +def _problem(status: int, title: str, **extra: Any) -> JSONResponse: + payload: dict[str, Any] = {"error": title, "status": status} + payload.update({k: v for k, v in extra.items() if v is not None}) + return JSONResponse(status_code=status, content=payload) + + +def register_exception_handlers(app: FastAPI) -> None: + @app.exception_handler(MissingCredentialsError) + async def _missing_credentials(request: Request, exc: MissingCredentialsError): + # Not an error worth a stack trace, but we still log it so missing-header + # problems are diagnosable. The credential VALUE is never logged. + logger.warning("Missing credentials for %s: %s", request.url.path, exc.message) + return _problem(401, "missing_credentials", detail=exc.message) + + @app.exception_handler(UpstreamError) + async def _upstream_error(request: Request, exc: UpstreamError): + logger.error( + "Upstream error on %s: %s (upstream_status=%s)", + request.url.path, + exc.message, + exc.upstream_status, + ) + return _problem( + exc.status, + "upstream_error", + detail=exc.message, + upstream_status=exc.upstream_status, + upstream_body=exc.body, + ) + + @app.exception_handler(Exception) + async def _unhandled(request: Request, exc: Exception): + # Last-resort handler: never leak a stack trace to the client, but always + # log it server-side so nothing fails silently. + logger.exception("Unhandled error on %s", request.url.path) + return _problem(500, "internal_error", detail=str(exc)) diff --git a/app/logging_config.py b/app/logging_config.py new file mode 100644 index 0000000..514f989 --- /dev/null +++ b/app/logging_config.py @@ -0,0 +1,30 @@ +"""Centralized logging setup. + +Project rule (see memory ``no-silent-failures``): every error or unexpected +state must reach the log. Never swallow an exception silently. Secrets +(tokens, service-account keys, sessions) must NEVER be logged. +""" +import logging +import os + +_LEVEL = os.getenv("LOG_LEVEL", "INFO").upper() + +_configured = False + + +def configure_logging() -> None: + """Configure root logging once. Safe to call multiple times.""" + global _configured + if _configured: + return + logging.basicConfig( + level=_LEVEL, + format="%(asctime)s %(levelname)s %(name)s %(message)s", + ) + _configured = True + + +def get_logger(name: str) -> logging.Logger: + """Return a module logger. Use ``get_logger(__name__)``.""" + configure_logging() + return logging.getLogger(name) diff --git a/app/main.py b/app/main.py index 33e388b..8a7d60b 100644 --- a/app/main.py +++ b/app/main.py @@ -1,25 +1,52 @@ +"""analytics - stateless API proxy for Google Analytics (GA4) and Sklik. + +Runs behind the AppFactory Caddy reverse proxy at /apps/. ROOT_PATH is +injected as an env var; FastAPI's ``root_path`` makes Swagger UI and the OpenAPI +``servers`` use the proxy prefix so "Try it out" hits /apps//... . + +The service stores no secrets. Every credential is supplied per request in an +X- header and used only to talk to the upstream API (see AGENTS.md and +``app.credentials``). +""" import os + from fastapi import FastAPI -APP_NAME = os.getenv("APP_NAME", "analytics") -APP_VERSION = os.getenv("APP_VERSION", "1.0.0") +from . import config +from .errors import register_exception_handlers +from .logging_config import get_logger +from .routers import ga_admin, ga_data, meta, sklik + +logger = get_logger(__name__) + ROOT_PATH = os.getenv("ROOT_PATH", "") +DESCRIPTION = """ +Stateless proxy exposing **Google Analytics 4** and **Sklik** (Seznam) APIs. + +All credentials are passed per request as `X-` headers (never stored): + +* **Google Analytics** — `X-GA-Access-Token` (preferred) or `X-GA-Credentials` + (base64 service-account JSON). Optional `X-GA-Quota-Project`. +* **Sklik** — `X-Sklik-Token`. Optional `X-Sklik-User-Id` for managed accounts. + +See `documentation/` in the repository for details. +""".strip() + app = FastAPI( - title=APP_NAME, - version=APP_VERSION, - root_path=ROOT_PATH + title=config.APP_NAME, + version=config.APP_VERSION, + description=DESCRIPTION, + root_path=ROOT_PATH, ) -@app.get("/health") -def health(): - return {"status": "ok"} +register_exception_handlers(app) -@app.get("/version") -def version(): - return { - "app": APP_NAME, - "version": APP_VERSION, - "language": "python", - "root_path": ROOT_PATH - } +app.include_router(meta.router) +app.include_router(ga_data.router) +app.include_router(ga_admin.router) +app.include_router(sklik.router) + +logger.info( + "analytics started (version=%s, root_path=%r)", config.APP_VERSION, ROOT_PATH +) diff --git a/app/routers/__init__.py b/app/routers/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/app/routers/ga_admin.py b/app/routers/ga_admin.py new file mode 100644 index 0000000..cc889f7 --- /dev/null +++ b/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 diff --git a/app/routers/ga_data.py b/app/routers/ga_data.py new file mode 100644 index 0000000..49938aa --- /dev/null +++ b/app/routers/ga_data.py @@ -0,0 +1,132 @@ +"""Google Analytics 4 - Data API (reporting). + +Thin passthrough: request bodies are the GA4 Data API request objects and +responses are returned as-is. See +https://developers.google.com/analytics/devguides/reporting/data/v1/rest + +Credentials: X-GA-Access-Token (preferred) or X-GA-Credentials. See +``app.credentials.get_ga_credentials``. +""" +from __future__ import annotations + +from typing import Any + +from fastapi import APIRouter, Body, Depends, Path + +from ..clients.ga_client import GoogleAnalyticsClient +from ..credentials import GaCredentials, get_ga_credentials + +router = APIRouter(prefix="/ga/data", tags=["google-analytics: data"]) + +# Reused OpenAPI example for report request bodies. +_RUN_REPORT_EXAMPLE = { + "dateRanges": [{"startDate": "7daysAgo", "endDate": "today"}], + "dimensions": [{"name": "country"}], + "metrics": [{"name": "activeUsers"}], +} + + +def _property(property_id: str) -> str: + # Accept both "123456789" and "properties/123456789". + pid = property_id.strip() + return pid if pid.startswith("properties/") else f"properties/{pid}" + + +@router.post( + "/properties/{property_id}/runReport", + summary="Run a GA4 report", +) +async def run_report( + property_id: str = Path(..., description="GA4 property id, e.g. 123456789"), + body: dict[str, Any] = Body(..., examples=[_RUN_REPORT_EXAMPLE]), + creds: GaCredentials = Depends(get_ga_credentials), +) -> Any: + client = GoogleAnalyticsClient(creds) + return await client.data_post(f"/{_property(property_id)}:runReport", body) + + +@router.post( + "/properties/{property_id}/runPivotReport", + summary="Run a GA4 pivot report", +) +async def run_pivot_report( + property_id: str = Path(..., description="GA4 property id"), + body: dict[str, Any] = Body(...), + creds: GaCredentials = Depends(get_ga_credentials), +) -> Any: + client = GoogleAnalyticsClient(creds) + return await client.data_post( + f"/{_property(property_id)}:runPivotReport", body + ) + + +@router.post( + "/properties/{property_id}/batchRunReports", + summary="Run up to 5 GA4 reports in one call", +) +async def batch_run_reports( + property_id: str = Path(..., description="GA4 property id"), + body: dict[str, Any] = Body(...), + creds: GaCredentials = Depends(get_ga_credentials), +) -> Any: + client = GoogleAnalyticsClient(creds) + return await client.data_post( + f"/{_property(property_id)}:batchRunReports", body + ) + + +@router.post( + "/properties/{property_id}/batchRunPivotReports", + summary="Run up to 5 GA4 pivot reports in one call", +) +async def batch_run_pivot_reports( + property_id: str = Path(..., description="GA4 property id"), + body: dict[str, Any] = Body(...), + creds: GaCredentials = Depends(get_ga_credentials), +) -> Any: + client = GoogleAnalyticsClient(creds) + return await client.data_post( + f"/{_property(property_id)}:batchRunPivotReports", body + ) + + +@router.post( + "/properties/{property_id}/runRealtimeReport", + summary="Run a GA4 realtime report", +) +async def run_realtime_report( + property_id: str = Path(..., description="GA4 property id"), + body: dict[str, Any] = Body(...), + creds: GaCredentials = Depends(get_ga_credentials), +) -> Any: + client = GoogleAnalyticsClient(creds) + return await client.data_post( + f"/{_property(property_id)}:runRealtimeReport", body + ) + + +@router.post( + "/properties/{property_id}/checkCompatibility", + summary="Check dimension/metric compatibility for a GA4 report", +) +async def check_compatibility( + property_id: str = Path(..., description="GA4 property id"), + body: dict[str, Any] = Body(...), + creds: GaCredentials = Depends(get_ga_credentials), +) -> Any: + client = GoogleAnalyticsClient(creds) + return await client.data_post( + f"/{_property(property_id)}:checkCompatibility", body + ) + + +@router.get( + "/properties/{property_id}/metadata", + summary="List available GA4 dimensions and metrics for a property", +) +async def get_metadata( + property_id: str = Path(..., description="GA4 property id"), + creds: GaCredentials = Depends(get_ga_credentials), +) -> Any: + client = GoogleAnalyticsClient(creds) + return await client.data_get(f"/{_property(property_id)}/metadata") diff --git a/app/routers/meta.py b/app/routers/meta.py new file mode 100644 index 0000000..a44b1d3 --- /dev/null +++ b/app/routers/meta.py @@ -0,0 +1,23 @@ +"""Infrastructure endpoints required by AppFactory. No credentials needed.""" +from fastapi import APIRouter + +from .. import config + +router = APIRouter(tags=["meta"]) + + +@router.get("/health", summary="Liveness/readiness probe") +def health() -> dict: + """Return 200 while the app can serve traffic. Used by AppFactory monitoring.""" + return {"status": "ok"} + + +@router.get("/version", summary="Service version and build info") +def version() -> dict: + return { + "app": config.APP_NAME, + "version": config.APP_VERSION, + "language": "python", + "root_path": config.ROOT_PATH, + "integrations": ["google-analytics", "sklik"], + } diff --git a/app/routers/sklik.py b/app/routers/sklik.py new file mode 100644 index 0000000..056ff65 --- /dev/null +++ b/app/routers/sklik.py @@ -0,0 +1,131 @@ +"""Sklik (Seznam) - JSON-RPC proxy. + +The proxy logs in with X-Sklik-Token per request (client.loginByToken) and then +performs the requested call, injecting the session for you. See +``app.clients.sklik_client`` and https://api.sklik.cz/drak/ for methods. + +Credentials: X-Sklik-Token (required). +""" +from __future__ import annotations + +from typing import Any + +from fastapi import APIRouter, Body, Depends, Header, Path + +from ..clients.sklik_client import SklikClient +from ..credentials import get_sklik_token +from ..errors import UpstreamError + +router = APIRouter(prefix="/sklik", tags=["sklik"]) + +# Entities that expose createReport/readReport for the report helper. +_REPORT_ENTITIES = { + "campaigns", + "groups", + "ads", + "keywords", + "queries", + "sitelinks", + "productSets", + "banners", +} + +_REPORT_EXAMPLE = [ + {"dateFrom": "2026-06-01", "dateTo": "2026-06-18", "statGranularity": "daily"}, + {"statGranularity": "daily"}, +] + + +def _optional_user_id( + x_sklik_user_id: str | None = Header( + default=None, + alias="X-Sklik-User-Id", + description="Optional managed account id (userId) to act on behalf of, " + "for agency/MCC access.", + ), +) -> int | None: + raw = (x_sklik_user_id or "").strip() + if not raw: + return None + try: + return int(raw) + except ValueError as exc: + raise UpstreamError( + "X-Sklik-User-Id must be an integer.", status=400 + ) from exc + + +@router.post("/login", summary="Verify the Sklik token (client.loginByToken)") +async def login( + token: str = Depends(get_sklik_token), + user_id: int | None = Depends(_optional_user_id), +) -> dict: + """Check the token works. The session itself is internal and not returned.""" + async with SklikClient(token, user_id=user_id) as client: + payload = await client.login() + return { + "valid": True, + "status": payload.get("status"), + "statusMessage": payload.get("statusMessage"), + } + + +@router.get("/limits", summary="API limits and quota (api.limits)") +async def limits( + token: str = Depends(get_sklik_token), + user_id: int | None = Depends(_optional_user_id), +) -> Any: + async with SklikClient(token, user_id=user_id) as client: + return await client.call("api.limits") + + +@router.post( + "/report/{entity}", + summary="Create and read a Sklik stats report (createReport + readReport)", +) +async def report( + entity: str = Path( + ..., + description="Entity to report on: " + + ", ".join(sorted(_REPORT_ENTITIES)), + ), + body: list[Any] = Body( + ..., + examples=[_REPORT_EXAMPLE], + description="Arguments for {entity}.createReport (restriction filter and " + "optional display options). The session is injected automatically.", + ), + token: str = Depends(get_sklik_token), + user_id: int | None = Depends(_optional_user_id), +) -> Any: + if entity not in _REPORT_ENTITIES: + raise UpstreamError( + f"Unsupported report entity '{entity}'. Allowed: " + + ", ".join(sorted(_REPORT_ENTITIES)), + status=400, + ) + async with SklikClient(token, user_id=user_id) as client: + return await client.fetch_report(entity, body) + + +@router.post( + "/rpc/{method}", + summary="Generic authenticated Sklik call (any method)", +) +async def rpc( + method: str = Path( + ..., + description="Sklik method name, e.g. campaigns.list, groups.list, " + "ads.list, api.limits. (client.loginByToken is managed by the proxy.)", + ), + args: list[Any] = Body( + default=[], + description="Positional arguments AFTER the session struct (which the " + "proxy injects as the first argument). Example for campaigns.list: " + '[{"statuses": ["active"]}, {"displayColumns": ["id","name"]}]', + ), + token: str = Depends(get_sklik_token), + user_id: int | None = Depends(_optional_user_id), +) -> Any: + async with SklikClient(token, user_id=user_id) as client: + return await client.call(method, args) diff --git a/documentation/google-analytics.md b/documentation/google-analytics.md new file mode 100644 index 0000000..f15786e --- /dev/null +++ b/documentation/google-analytics.md @@ -0,0 +1,84 @@ +# Google Analytics (GA4) + +Proxy over the GA4 **Data API** (`analyticsdata.googleapis.com/v1beta`) and +**Admin API** (`analyticsadmin.googleapis.com/v1beta`). + +## Credentials + +Token has precedence over the service account: + +| Header | Meaning | +| --- | --- | +| `X-GA-Access-Token` | Ready OAuth2 access token, used directly as `Authorization: Bearer`. | +| `X-GA-Credentials` | **Base64** of a Google service-account JSON key. The proxy mints a short-lived token (scope `https://www.googleapis.com/auth/analytics.readonly`) via `google-auth` and caches it in memory until ~60 s before expiry. | +| `X-GA-Quota-Project` | Optional GCP project id → upstream `x-goog-user-project`. | + +At least one of `X-GA-Access-Token` / `X-GA-Credentials` is required (otherwise +`401 missing_credentials`). + +The service account (or token) must have access to the GA4 property — add its +`client_email` as a viewer in GA Admin → Property Access Management. + +> Encoding the key: `base64 -w0 service-account.json` (Linux) or +> `[Convert]::ToBase64String([IO.File]::ReadAllBytes("service-account.json"))` +> (PowerShell). + +## Data API endpoints + +`property_id` may be the bare number (`123456789`) or `properties/123456789`. + +| Method | Path | Upstream | +| --- | --- | --- | +| POST | `/ga/data/properties/{id}/runReport` | `:runReport` | +| POST | `/ga/data/properties/{id}/runPivotReport` | `:runPivotReport` | +| POST | `/ga/data/properties/{id}/batchRunReports` | `:batchRunReports` | +| POST | `/ga/data/properties/{id}/batchRunPivotReports` | `:batchRunPivotReports` | +| POST | `/ga/data/properties/{id}/runRealtimeReport` | `:runRealtimeReport` | +| POST | `/ga/data/properties/{id}/checkCompatibility` | `:checkCompatibility` | +| GET | `/ga/data/properties/{id}/metadata` | `/metadata` | + +The POST body is the GA4 request object, forwarded unchanged. Example +`runReport` body: + +```json +{ + "dateRanges": [{ "startDate": "7daysAgo", "endDate": "today" }], + "dimensions": [{ "name": "country" }], + "metrics": [{ "name": "activeUsers" }] +} +``` + +## Admin API endpoints (read) + +| Method | Path | Notes | +| --- | --- | --- | +| GET | `/ga/admin/accounts` | `pageSize`, `pageToken` | +| GET | `/ga/admin/accountSummaries` | accounts + their properties | +| GET | `/ga/admin/properties?accountId=123` | builds `filter=parent:accounts/123` | +| GET | `/ga/admin/properties/{id}` | single property | +| GET | `/ga/admin/properties/{id}/dataStreams` | data streams | + +## Errors + +`UpstreamError` is returned as JSON with the upstream status and body: + +```json +{ + "error": "upstream_error", + "status": 403, + "detail": "User does not have sufficient permissions for this property.", + "upstream_status": 403, + "upstream_body": { "error": { "code": 403, "status": "PERMISSION_DENIED" } } +} +``` + +Timeouts → `504`, unreachable/transport → `502`, token minting failure → `401`. + +## curl example + +```bash +curl -X POST "https://services.csbot.cz/apps/analytics/ga/data/properties/123456789/runReport" \ + -H "X-GA-Access-Token: ya29...." \ + -H "Content-Type: application/json" \ + -d '{"dateRanges":[{"startDate":"7daysAgo","endDate":"today"}],"metrics":[{"name":"activeUsers"}]}' +``` diff --git a/documentation/overview.md b/documentation/overview.md new file mode 100644 index 0000000..56c61f2 --- /dev/null +++ b/documentation/overview.md @@ -0,0 +1,71 @@ +# analytics — overview + +A stateless multi-tenant API proxy exposing two upstream services under one +FastAPI app: + +1. **Google Analytics 4** — Data API (reporting) + Admin API (read). +2. **Sklik** (Seznam) — Drak JSON-RPC API. + +The structure mirrors the sibling `idoklad` / `csob` services (config→env, +credentials→headers, client per upstream, routers, central exception handling, +Swagger at `/docs`), adapted to Python/FastAPI. + +## Design principles + +- **Stateless / no stored secrets.** Credentials arrive per request in `X-` + headers and are used only to call the upstream. Nothing is persisted; the + only in-memory state is a short-lived GA access-token cache (see below). +- **Thin passthrough.** GA request/response bodies and most Sklik calls are + forwarded as-is, so callers keep the full upstream API surface. Only + authentication, base URL and error mapping are added. +- **No silent failures.** Every error is logged (never the secret values) and + surfaced as JSON. Upstream errors preserve the upstream status and body. + +## Layout + +``` +app/ + config.py env-driven config (base URLs, scope, timeout) — no secrets + logging_config.py get_logger(); secrets are never logged + errors.py MissingCredentialsError, UpstreamError + handlers + credentials.py X- header dependencies (GA + Sklik) + clients/ + ga_client.py GA Data/Admin HTTP client + service-account token minting + sklik_client.py Sklik JSON-RPC client (login + session + report paging) + routers/ + meta.py /health, /version + ga_data.py /ga/data/... + ga_admin.py /ga/admin/... + sklik.py /sklik/... + main.py app factory, root_path, router + handler registration +``` + +## Reverse proxy + +`ROOT_PATH` (e.g. `/apps/analytics`) is passed to FastAPI's `root_path`, so the +OpenAPI `servers` entry and Swagger "Try it out" use the public prefix. Internal +routes are unprefixed (Caddy `handle_path` strips the prefix). + +## Authentication summary + +| Upstream | Header(s) | Behaviour | +| --- | --- | --- | +| Google Analytics | `X-GA-Access-Token` **or** `X-GA-Credentials` (+ `X-GA-Quota-Project`) | Token used directly; else a token is minted from the base64 service-account JSON (scope `analytics.readonly`) and cached in memory until ~60 s before expiry. | +| Sklik | `X-Sklik-Token` (+ `X-Sklik-User-Id`) | `client.loginByToken` per request → session injected into the call. | + +## Deliberately not wired + +- **GA Admin write operations** (create/update/delete properties, streams). The + requested scope is read-only (`analytics.readonly`); add `analytics.edit` and + endpoints if management is needed later. +- **Sklik header-credential encryption.** Same deferral as `idoklad`/`csob`: + header values are plaintext over TLS for now. +- **Sklik session reuse across requests** — the chosen model logs in per + request; a future `X-Sklik-Session` passthrough could save the login call. + +## Verification checklist (per AGENTS.md) + +- `/health` returns 200. +- `/docs` loads; `/openapi.json` `servers` contains the proxy prefix. +- New endpoints appear in Swagger with their `X-` headers in "Try it out". +- Secrets never appear in logs or source. diff --git a/documentation/sklik.md b/documentation/sklik.md new file mode 100644 index 0000000..0c193d0 --- /dev/null +++ b/documentation/sklik.md @@ -0,0 +1,84 @@ +# Sklik (Seznam) + +Proxy over the Sklik **Drak JSON API** +(`https://api.sklik.cz/drak/json/v5/{method}`). + +## Protocol (verified against seznam/api-examples) + +- HTTP `POST` to the base URL with the **method name appended** to the path. +- Body is a **JSON array** of positional arguments. +- `client.loginByToken` takes the API token and returns + `{"status":200,"session":"...","statusMessage":"OK"}`. +- Every authenticated method takes the user struct `{"session": ...}` (optionally + `"userId"`) as its **first** argument, followed by the method's own arguments. +- Every response is an object with `status` (HTTP-style), `statusMessage`, a + refreshed `session`, and method-specific data. `200`, `206` and `301` are + treated as success. + +The proxy performs `client.loginByToken` per request from `X-Sklik-Token` and +injects the session — callers never handle the session. + +## Credentials + +| Header | Required | Meaning | +| --- | --- | --- | +| `X-Sklik-Token` | yes | API token from Sklik → account settings → API. | +| `X-Sklik-User-Id` | no | Managed account `userId` (agency/MCC access). | + +Missing token → `401 missing_credentials`. Sklik business errors (invalid token, +access denied, bad arguments) are surfaced as `upstream_error` with the Sklik +`status` and full body. + +## Endpoints + +| Method | Path | Purpose | +| --- | --- | --- | +| POST | `/sklik/login` | Verify the token. Returns `{valid, status, statusMessage}` (no session). | +| GET | `/sklik/limits` | `api.limits` — quotas and the `statsDataLimit`. | +| POST | `/sklik/report/{entity}` | `createReport` + paged `readReport` for an entity. | +| POST | `/sklik/rpc/{method}` | Generic authenticated call to any method. | + +### Report helper + +`entity` ∈ `campaigns, groups, ads, keywords, queries, sitelinks, productSets, +banners`. Body = the arguments for `{entity}.createReport` (restriction filter + +optional display options). The proxy creates the report then pages through +`{entity}.readReport` (100 rows/page) and returns: + +```json +{ "reportId": "...", "totalCount": 1234, "returnedCount": 1234, "truncated": false, "report": [ ... ] } +``` + +Example body for `POST /sklik/report/campaigns`: + +```json +[ + { "dateFrom": "2026-06-01", "dateTo": "2026-06-18", "statGranularity": "daily" }, + { "statGranularity": "daily" } +] +``` + +### Generic RPC + +`POST /sklik/rpc/{method}` with a JSON-array body of the arguments **after** the +session struct (which the proxy injects). Examples: + +```bash +# List campaigns +curl -X POST ".../apps/analytics/sklik/rpc/campaigns.list" \ + -H "X-Sklik-Token: " -H "Content-Type: application/json" \ + -d '[{"statuses":["active"]}, {"displayColumns":["id","name","status"]}]' + +# Account info +curl -X POST ".../apps/analytics/sklik/rpc/client.get" \ + -H "X-Sklik-Token: " -H "Content-Type: application/json" -d '[]' +``` + +`client.loginByToken` cannot be called via `/sklik/rpc` — the proxy manages the +session (returns `400`). + +## Method reference + +Full method list: . Common ones: `client.get`, +`api.limits`, `campaigns.list`, `groups.list`, `ads.list`, `keywords.list`, +`*.createReport` / `*.readReport`. diff --git a/requirements.txt b/requirements.txt index 364e2ee..9f924f3 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,2 +1,4 @@ fastapi uvicorn[standard] +httpx +google-auth