first

app/routers/insights.py

@@ -0,0 +1,325 @@
+"""Meta Marketing API - insights (spend, impressions, clicks, conversions).
+
+Insights come in two flavours upstream and both are exposed here:
+
+  * **synchronous** - ``GET /{object_id}/insights``. Fine for one account or a
+    handful of campaigns over a short period.
+  * **asynchronous** - large reports (long date ranges, many breakdowns, whole
+    accounts at ad level) are run as a job on Meta's side: start the job, poll
+    its status, then read the result. Meta will reject or time out a sync call
+    that is too big, so anything sizeable belongs here.
+
+``POST /ads/insights/{object_id}/run`` wraps the whole async dance (start →
+poll → read) in one call for callers that just want the numbers and can wait.
+
+``object_id`` is anything Meta can report on: an ad account (``act_123`` or
+``123``), a campaign id, an ad set id or an ad id.
+
+All read-only. Credentials: see ``app.credentials.get_meta_credentials``.
+"""
+from __future__ import annotations
+
+import asyncio
+import time
+from typing import Any
+
+from fastapi import APIRouter, Depends, Path, Query
+
+from .. import config
+from ..clients.graph import MetaGraphClient
+from ..credentials import MetaCredentials, get_meta_credentials
+from ..errors import UpstreamError
+from ..logging_config import get_logger
+
+logger = get_logger(__name__)
+
+router = APIRouter(prefix="/ads/insights", tags=["meta-ads: insights"])
+
+# Level-agnostic default metrics: valid whether the caller reports at account,
+# campaign, adset or ad level. Name/id fields are level-specific (asking for
+# ad_id at campaign level is an upstream error), so callers add those via
+# `fields` together with `level` - see documentation/meta-ads.md.
+_DEFAULT_FIELDS = (
+    "spend,impressions,clicks,ctr,cpc,cpm,reach,frequency,"
+    "actions,action_values,date_start,date_stop"
+)
+
+# Terminal states of an async insights job.
+_JOB_DONE = "Job Completed"
+_JOB_FAILED = {"Job Failed", "Job Skipped"}
+
+
+def _client(creds: MetaCredentials) -> MetaGraphClient:
+    return MetaGraphClient(creds, service_name="Meta Marketing API")
+
+
+def _object(object_id: str) -> str:
+    """Normalize the reporting object id.
+
+    A bare numeric ad account id is ambiguous upstream, so a digits-only id that
+    the caller labelled as an account still needs the act_ prefix; campaign /
+    adset / ad ids are passed through untouched.
+    """
+    value = object_id.strip()
+    return value
+
+
+def _insight_params(
+    fields: str | None,
+    level: str | None,
+    date_preset: str | None,
+    time_range: str | None,
+    time_increment: str | None,
+    breakdowns: str | None,
+    action_breakdowns: str | None,
+    filtering: str | None,
+    sort: str | None,
+    limit: int | None = None,
+    after: str | None = None,
+) -> dict[str, Any]:
+    return {
+        "fields": fields,
+        "level": level,
+        "date_preset": date_preset,
+        "time_range": time_range,
+        "time_increment": time_increment,
+        "breakdowns": breakdowns,
+        "action_breakdowns": action_breakdowns,
+        "filtering": filtering,
+        "sort": sort,
+        "limit": limit,
+        "after": after,
+    }
+
+
+# Shared Query definitions so the sync and async endpoints stay in step.
+_Q_FIELDS = Query(_DEFAULT_FIELDS, description="Comma-separated insight fields/metrics.")
+_Q_LEVEL = Query(
+    None, description="Aggregation level: account, campaign, adset or ad."
+)
+_Q_DATE_PRESET = Query(
+    None,
+    description="Relative period, e.g. today, yesterday, last_7d, last_30d, "
+    "this_month, last_month, maximum. Ignored if time_range is given.",
+)
+_Q_TIME_RANGE = Query(
+    None,
+    description='Absolute period as JSON: {"since":"2026-06-01","until":"2026-06-30"}.',
+)
+_Q_TIME_INCREMENT = Query(
+    None,
+    description="Row granularity: number of days (e.g. 1 = daily), 'monthly', "
+    "or 'all_days' for a single summed row.",
+)
+_Q_BREAKDOWNS = Query(
+    None,
+    description="Comma-separated breakdowns, e.g. age,gender or "
+    "publisher_platform,platform_position or country.",
+)
+_Q_ACTION_BREAKDOWNS = Query(
+    None, description="Comma-separated action breakdowns, e.g. action_type."
+)
+_Q_FILTERING = Query(
+    None,
+    description='JSON array of filters, e.g. '
+    '[{"field":"spend","operator":"GREATER_THAN","value":100}].',
+)
+_Q_SORT = Query(None, description="Sort spec, e.g. spend_descending.")
+
+
+@router.get("/{object_id}", summary="Insights, synchronous (small reports)")
+async def insights(
+    object_id: str = Path(
+        ...,
+        description="Ad account (act_123), campaign, ad set or ad id to report on.",
+    ),
+    fields: str = _Q_FIELDS,
+    level: str | None = _Q_LEVEL,
+    date_preset: str | None = _Q_DATE_PRESET,
+    time_range: str | None = _Q_TIME_RANGE,
+    time_increment: str | None = _Q_TIME_INCREMENT,
+    breakdowns: str | None = _Q_BREAKDOWNS,
+    action_breakdowns: str | None = _Q_ACTION_BREAKDOWNS,
+    filtering: str | None = _Q_FILTERING,
+    sort: str | None = _Q_SORT,
+    limit: int | None = Query(None, ge=1, le=500),
+    after: str | None = Query(None, description="Paging cursor."),
+    all_pages: bool = Query(
+        True,
+        description="Follow paging.next and return every page (default; capped "
+        "by META_MAX_PAGES, the result says truncated:true if the cap is hit). "
+        "Set false for a single raw page. For big reports prefer the async "
+        "endpoints below - paging a huge sync report is what Meta rejects.",
+    ),
+    creds: MetaCredentials = Depends(get_meta_credentials),
+) -> Any:
+    params = _insight_params(
+        fields,
+        level,
+        date_preset,
+        time_range,
+        time_increment,
+        breakdowns,
+        action_breakdowns,
+        filtering,
+        sort,
+        limit,
+        after,
+    )
+    path = f"{_object(object_id)}/insights"
+    client = _client(creds)
+    if all_pages:
+        return await client.get_all_pages(path, params)
+    return await client.get(path, params)
+
+
+@router.post("/{object_id}/jobs", summary="Start an async insights job")
+async def start_job(
+    object_id: str = Path(..., description="Ad account, campaign, ad set or ad id."),
+    fields: str = _Q_FIELDS,
+    level: str | None = _Q_LEVEL,
+    date_preset: str | None = _Q_DATE_PRESET,
+    time_range: str | None = _Q_TIME_RANGE,
+    time_increment: str | None = _Q_TIME_INCREMENT,
+    breakdowns: str | None = _Q_BREAKDOWNS,
+    action_breakdowns: str | None = _Q_ACTION_BREAKDOWNS,
+    filtering: str | None = _Q_FILTERING,
+    sort: str | None = _Q_SORT,
+    creds: MetaCredentials = Depends(get_meta_credentials),
+) -> Any:
+    """Queue the report on Meta's side. Returns ``report_run_id`` to poll."""
+    form = _insight_params(
+        fields,
+        level,
+        date_preset,
+        time_range,
+        time_increment,
+        breakdowns,
+        action_breakdowns,
+        filtering,
+        sort,
+    )
+    return await _client(creds).post(f"{_object(object_id)}/insights", form)
+
+
+@router.get("/jobs/{report_run_id}", summary="Async insights job status")
+async def job_status(
+    report_run_id: str = Path(..., description="report_run_id returned when starting the job."),
+    creds: MetaCredentials = Depends(get_meta_credentials),
+) -> Any:
+    return await _client(creds).get(
+        report_run_id.strip(),
+        {
+            "fields": "async_status,async_percent_completion,date_start,date_stop,"
+            "time_completed,emails"
+        },
+    )
+
+
+@router.get("/jobs/{report_run_id}/results", summary="Read a finished async job")
+async def job_results(
+    report_run_id: str = Path(...),
+    limit: int | None = Query(None, ge=1, le=500),
+    after: str | None = Query(None),
+    all_pages: bool = Query(True, description="Follow paging.next across result pages."),
+    creds: MetaCredentials = Depends(get_meta_credentials),
+) -> Any:
+    path = f"{report_run_id.strip()}/insights"
+    params = {"limit": limit, "after": after}
+    client = _client(creds)
+    if all_pages:
+        return await client.get_all_pages(path, params)
+    return await client.get(path, params)
+
+
+@router.post(
+    "/{object_id}/run",
+    summary="Async insights: start, wait for completion and return the rows",
+)
+async def run_and_wait(
+    object_id: str = Path(..., description="Ad account, campaign, ad set or ad id."),
+    fields: str = _Q_FIELDS,
+    level: str | None = _Q_LEVEL,
+    date_preset: str | None = _Q_DATE_PRESET,
+    time_range: str | None = _Q_TIME_RANGE,
+    time_increment: str | None = _Q_TIME_INCREMENT,
+    breakdowns: str | None = _Q_BREAKDOWNS,
+    action_breakdowns: str | None = _Q_ACTION_BREAKDOWNS,
+    filtering: str | None = _Q_FILTERING,
+    sort: str | None = _Q_SORT,
+    max_wait_seconds: float | None = Query(
+        None,
+        ge=1,
+        description="Override the wait budget (default META_ASYNC_MAX_WAIT_SECONDS).",
+    ),
+    creds: MetaCredentials = Depends(get_meta_credentials),
+) -> Any:
+    """Convenience wrapper over start → poll → read.
+
+    On timeout the job is NOT cancelled - the response carries the
+    ``report_run_id`` so the caller can keep polling ``/jobs/{id}`` instead of
+    losing the work already done upstream.
+    """
+    client = _client(creds)
+    form = _insight_params(
+        fields,
+        level,
+        date_preset,
+        time_range,
+        time_increment,
+        breakdowns,
+        action_breakdowns,
+        filtering,
+        sort,
+    )
+    started = await client.post(f"{_object(object_id)}/insights", form)
+    run_id = (started or {}).get("report_run_id") if isinstance(started, dict) else None
+    if not run_id:
+        raise UpstreamError(
+            "Meta did not return a report_run_id for the async insights job.",
+            status=502,
+            body=started,
+        )
+
+    budget = max_wait_seconds or config.META_ASYNC_MAX_WAIT_SECONDS
+    deadline = time.monotonic() + budget
+    status_fields = {"fields": "async_status,async_percent_completion"}
+
+    while True:
+        status = await client.get(str(run_id), status_fields)
+        async_status = (status or {}).get("async_status") if isinstance(status, dict) else None
+
+        if async_status == _JOB_DONE:
+            results = await client.get_all_pages(f"{run_id}/insights", None)
+            if isinstance(results, dict):
+                results["report_run_id"] = run_id
+            return results
+
+        if async_status in _JOB_FAILED:
+            raise UpstreamError(
+                f"Async insights job {run_id} ended with status '{async_status}'.",
+                status=502,
+                body=status,
+            )
+
+        if time.monotonic() >= deadline:
+            # Not an error upstream - the job is still running. Say so clearly
+            # and hand back the id rather than failing silently or hanging.
+            logger.warning(
+                "Async insights job %s still running after %.0fs; returning id to caller.",
+                run_id,
+                budget,
+            )
+            return {
+                "report_run_id": run_id,
+                "completed": False,
+                "async_status": async_status,
+                "async_percent_completion": (status or {}).get("async_percent_completion"),
+                "detail": (
+                    f"Job did not finish within {budget:.0f}s. It is still running "
+                    f"upstream - poll /ads/insights/jobs/{run_id} and then read "
+                    f"/ads/insights/jobs/{run_id}/results."
+                ),
+            }
+
+        await asyncio.sleep(config.META_ASYNC_POLL_INTERVAL_SECONDS)