Google Ad Manager

GAM_NETWORK_CODE=12345678
GAM_KEY_FILE=service-account.json

-- models/raw/gam_report.sql
-- @kind: append
-- @incremental: DATE
-- @incremental_initial: 20260401

SELECT
    AD_UNIT_NAME,
    DATE,
    ORDER_NAME,
    LINE_ITEM_NAME,
    AD_SERVER_IMPRESSIONS::BIGINT AS AD_SERVER_IMPRESSIONS,
    AD_SERVER_CLICKS::BIGINT AS AD_SERVER_CLICKS,
    AD_SERVER_CTR::DOUBLE AS AD_SERVER_CTR,
    AD_SERVER_REVENUE::DOUBLE AS AD_SERVER_REVENUE
FROM gam_report()

-- models/staging/gam_report.sql
-- @kind: table
-- @constraint: not_null(report_date)

SELECT
    AD_UNIT_NAME AS ad_unit,
    strptime(DATE, '%Y%m%d')::DATE AS report_date,
    ORDER_NAME AS order_name,
    LINE_ITEM_NAME AS line_item,
    AD_SERVER_IMPRESSIONS::BIGINT AS impressions,
    AD_SERVER_CLICKS::BIGINT AS clicks,
    AD_SERVER_CTR::DOUBLE AS ctr,
    AD_SERVER_REVENUE::DOUBLE AS revenue
FROM raw.gam_report

ondatrasql run

ondatrasql run raw.gam_report

FROM gam_report('{"custom_dimensions": [11678108], "currency": "USD"}')

FROM gam_report('{"currency": "SEK", "timezone": "Europe/Stockholm"}')

FROM gam_report('{"filters": [{"field": "AD_UNIT_NAME", "operation": "EQUALS", "values": ["Mobile_1_320x320"]}]}')

FROM gam_report('{"custom_dimensions": [11678108]}')

-- models/raw/gam_with_custom.sql
-- @kind: append
-- @incremental: DATE
-- @incremental_initial: 20260420

SELECT
    AD_UNIT_NAME,
    DATE,
    CUSTOM_DIMENSION_0_VALUE_ID,
    CUSTOM_DIMENSION_0_VALUE,
    AD_SERVER_IMPRESSIONS::BIGINT AS AD_SERVER_IMPRESSIONS,
    AD_SERVER_REVENUE::DOUBLE AS AD_SERVER_REVENUE
FROM gam_report('{"custom_dimensions": [11678108]}')

SELECT DATE, AD_SERVER_IMPRESSIONS::BIGINT, AD_SERVER_REVENUE::DOUBLE
FROM gam_report()

SELECT
    AD_UNIT_NAME, DATE, COUNTRY_NAME,
    AD_SERVER_IMPRESSIONS::BIGINT, AD_SERVER_REVENUE::DOUBLE
FROM gam_report()

-- models/mart/daily_revenue.sql
-- @kind: table

SELECT
    report_date,
    SUM(impressions) AS total_impressions,
    SUM(clicks) AS total_clicks,
    SUM(revenue) AS total_revenue,
    CASE WHEN SUM(impressions) > 0
        THEN SUM(revenue) / SUM(impressions) * 1000
        ELSE 0
    END AS rpm
FROM staging.gam_report
GROUP BY report_date
ORDER BY report_date

RESERVED_COLUMNS = ()

API = {
    "base_url": "https://admanager.googleapis.com",
    "auth": {
        "service_account": {"env": "GAM_KEY_FILE"},
        "scope": "https://www.googleapis.com/auth/admanager",
    },
    "timeout": 120,
    "retry": 3,
    "backoff": 2,
    "fetch": {
        "args": ["options"],
        "page_size": 10000,
        "supported_kinds": ["append"],
        "async": True,
        "poll_interval": "5s",
        "poll_timeout": "5m",
        "poll_backoff": 2,
    },
}

def submit(options="", columns=[], is_backfill=True, last_value="", initial_value=""):
    base = "/v1/networks/" + env.get("GAM_NETWORK_CODE")
    one_day = time.parse_duration("24h")
    yesterday = (time.now() - one_day).format("2006-01-02")

    if is_backfill:
        start_date = _to_iso_date(initial_value)
    else:
        start_date = _next_day(_to_iso_date(last_value))

    if start_date > yesterday:
        abort()

    # Normalized types: string = dimension, anything else = metric
    dims = [c["name"] for c in columns if c.get("type", "string") == "string"]
    mets = [c["name"] for c in columns if c.get("type", "string") != "string"]

    s = time.parse_time(start_date + "T00:00:00Z")
    e = time.parse_time(yesterday + "T00:00:00Z")

    opts = json.decode(options) if options else {}

    report_def = {
        "dimensions": dims,
        "metrics": mets,
        "reportType": opts.get("report_type", "HISTORICAL"),
        "dateRange": {"fixed": {
            "startDate": {"year": s.year, "month": s.month, "day": s.day},
            "endDate":   {"year": e.year, "month": e.month, "day": e.day},
        }},
    }

    if "custom_dimensions" in opts:
        report_def["customDimensionKeyIds"] = [str(id) for id in opts["custom_dimensions"]]
    if "cms_metadata" in opts:
        report_def["cmsMetadataDimensionKeyIds"] = [str(id) for id in opts["cms_metadata"]]
    if "ekv_dimensions" in opts:
        report_def["ekvDimensionKeyIds"] = [str(id) for id in opts["ekv_dimensions"]]
    if "line_item_custom_fields" in opts:
        report_def["lineItemCustomFieldIds"] = [str(id) for id in opts["line_item_custom_fields"]]
    if "order_custom_fields" in opts:
        report_def["orderCustomFieldIds"] = [str(id) for id in opts["order_custom_fields"]]
    if "creative_custom_fields" in opts:
        report_def["creativeCustomFieldIds"] = [str(id) for id in opts["creative_custom_fields"]]
    if "currency" in opts:
        report_def["currencyCode"] = opts["currency"]
    if "timezone" in opts:
        report_def["timeZone"] = opts["timezone"]
        report_def["timeZoneSource"] = "PROVIDED"
    if "filters" in opts:
        report_def["filters"] = opts["filters"]
    if "flags" in opts:
        report_def["flags"] = opts["flags"]
    if "expanded_compatibility" in opts:
        report_def["expandedCompatibility"] = opts["expanded_compatibility"]

    # Create report
    resp = http.post(base + "/reports", json={
        "reportDefinition": report_def,
        "visibility": "HIDDEN",
    })
    if not resp.ok:
        fail("create report: " + resp.text)

    # Run report
    resp = http.post("/v1/" + resp.json["name"] + ":run", json={})
    if not resp.ok:
        fail("run report: " + resp.text)

    return {"operation": resp.json["name"], "dims": dims, "mets": mets}


def check(job_ref):
    resp = http.get("/v1/" + job_ref["operation"])
    if not resp.ok:
        fail("poll report: " + resp.text)

    if not resp.json.get("done"):
        return None

    if "error" in resp.json:
        fail("report failed: " + json.encode(resp.json["error"]))

    result = resp.json["response"]["reportResult"]
    return {
        "fetch_url": "/v1/" + result + ":fetchRows",
        "dims": job_ref["dims"],
        "mets": job_ref["mets"],
    }


def fetch_result(result_ref, page):
    url = result_ref["fetch_url"] + "?pageSize=" + str(page.size)
    if page.cursor != None:
        url += "&pageToken=" + str(page.cursor)

    resp = http.get(url)
    if not resp.ok:
        fail("fetch rows: " + resp.text)

    dims = result_ref["dims"]
    mets = result_ref["mets"]
    rows = []

    for row in (resp.json.get("rows") or []):
        dv = row.get("dimensionValues") or []
        mv = (row.get("metricValueGroups") or [{}])[0].get("primaryValues") or []
        record = {}
        for i, name in enumerate(dims):
            if i < len(dv):
                record[name] = _extract_value(dv[i])
        for i, name in enumerate(mets):
            if i < len(mv):
                record[name] = _extract_value(mv[i])
        rows.append(record)

    return {"rows": rows, "next": resp.json.get("nextPageToken")}


def _extract_value(v):
    # GAM ReportValue: positional mapping guaranteed by API docs.
    # "The order of the dimension values is the same as the order
    # of the dimensions specified in the request."
    if "stringValue" in v:
        return v["stringValue"]
    if "intValue" in v:
        return v["intValue"]
    if "doubleValue" in v:
        return v["doubleValue"]
    if "boolValue" in v:
        return v["boolValue"]
    if "dateValue" in v:
        d = v["dateValue"]
        return "%04d-%02d-%02d" % (d["year"], d["month"], d["day"])
    fail("unknown GAM ReportValue type: " + json.encode(v))


def _to_iso_date(d):
    """Normalize to YYYY-MM-DD. Handles YYYYMMDD (GAM format) and YYYY-MM-DD."""
    if len(d) >= 8 and d[4] != "-":
        return d[:4] + "-" + d[4:6] + "-" + d[6:8]
    return d[:10]


def _next_day(iso_date):
    """Next calendar day. Uses UTC — no DST edge cases."""
    d = time.parse_time(iso_date + "T00:00:00Z")
    return (d + time.parse_duration("24h")).format("2006-01-02")