search a ads

app/clients/google.py

@@ -0,0 +1,190 @@
+"""Shared Google API client used by Analytics, Search Console and Google Ads.
+
+A single Google OAuth mechanism backs all three services - the only difference
+is the OAuth *scope* (carried on ``GoogleCredentials.scope``) and, for Google
+Ads, a couple of extra headers. Request/response bodies are forwarded as-is so
+callers keep the full upstream API surface; this module only handles
+authentication (Bearer token), the optional quota-project header and error
+mapping.
+
+Auth (token wins over service account):
+  * If an access token was supplied in the header, it is used directly.
+  * Otherwise a short-lived access token is minted from the service-account JSON
+    via google-auth for the requested scope and cached in-memory (keyed by
+    key id + email + scope) until shortly before it expires. 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 GoogleCredentials
+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"Service-account credentials are not usable: {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: GoogleCredentials) -> str:
+    if creds.access_token:
+        return creds.access_token
+
+    if creds.service_account_info is None:
+        # Credential dependencies guarantee one of the two, but be defensive.
+        raise MissingCredentialsError(
+            "No Google access token and no service-account credentials available."
+        )
+
+    scope = creds.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 GoogleApiClient:
+    """Authenticated HTTP client for any Google REST API.
+
+    ``extra_headers`` lets callers add upstream-specific headers (e.g. the
+    Google Ads ``developer-token`` / ``login-customer-id``).
+    """
+
+    def __init__(
+        self,
+        creds: GoogleCredentials,
+        extra_headers: dict[str, str] | None = None,
+        *,
+        service_name: str = "Google",
+    ) -> None:
+        self._creds = creds
+        self._extra_headers = extra_headers or {}
+        self._service_name = service_name
+
+    async def request(
+        self,
+        method: str,
+        url: 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
+        headers.update(self._extra_headers)
+
+        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(
+                f"{self._service_name} request timed out.", status=504
+            ) from exc
+        except httpx.HTTPError as exc:
+            raise UpstreamError(
+                f"{self._service_name} is unreachable: {exc}", status=502
+            ) from exc
+
+        return _parse_google_response(resp, self._service_name)
+
+    async def get(self, url: str, params: dict | None = None) -> Any:
+        return await self.request("GET", url, params=params)
+
+    async def post(self, url: str, json_body: Any) -> Any:
+        return await self.request("POST", url, json_body=json_body)
+
+
+def _parse_google_response(resp: httpx.Response, service_name: str) -> Any:
+    try:
+        payload = resp.json()
+    except ValueError:
+        payload = {"raw": resp.text}
+
+    if resp.is_success:
+        return payload
+
+    # Google returns {"error": {"code", "message", "status", ...}}; Google Ads
+    # streaming returns a list whose first element may carry the error.
+    message = f"{service_name} API error"
+    err_obj: Any = None
+    if isinstance(payload, dict):
+        err_obj = payload.get("error")
+    elif isinstance(payload, list) and payload and isinstance(payload[0], dict):
+        err_obj = payload[0].get("error")
+    if isinstance(err_obj, dict) and err_obj.get("message"):
+        message = err_obj["message"]
+
+    raise UpstreamError(
+        message,
+        status=502 if resp.status_code >= 500 else resp.status_code,
+        upstream_status=resp.status_code,
+        body=payload,
+    )