LinkedIn Ads ➡️ Snowflake, with 3-legged OAuth in the middle

TL/DR: LinkedIn Marketing APIs require 3-legged OAuth, which means a human in a browser. Scheduled pipelines don't have browsers. I store the refresh token in Snowflake.

I built a dlt pipeline to load LinkedIn Ads data into Snowflake. It runs on a schedule in dltHub Runtime, which spins up a fresh container, runs the job, then throws the container away. Ephemeral by design. No persistent disk. No place to hide long-lived credentials.

This is rarely an issue, but... LinkedIn's Marketing APIs require 3-legged OAuth. A full authorization-code flow with a real human clicking "Allow" in a browser.

Here's how I made it work.

3-legged OAuth, briefly

Most modern APIs accept a static API key or a 2-legged client credentials flow. Send a client_id + client_secret, get a token back, done. Great for headless pipelines.

LinkedIn Marketing APIs don't do that. They want a real human to log into LinkedIn, look at the app's requested scopes, and click "Allow". The dance:

1. App redirects user to LinkedIn's authorize URL with client_id, redirect_uri, and scope
2. User logs in (or is already logged in), clicks Allow
3. LinkedIn redirects back to redirect_uri with a one-time code
4. App POSTs that code to LinkedIn's token endpoint, gets back access_token + refresh_token
5. App uses access_token for API calls until it expires (~60 minutes)
6. When access_token expires, app exchanges refresh_token for a new one
7. Refresh token lasts ~12 months for Marketing APIs; when it expires, the human comes back

Step 6 is the key one for pipelines. As long as the refresh token is alive, no human is needed. The pipeline can mint a fresh access token on every run, automatically.

LinkedIn 3-Legged OAuth Flow - LinkedIn

Step-by-step guide for LinkedIn’s 3-legged OAuth flow.

Microsoft LearnDeeksha-ramesh

The container problem

dltHub has a scaffold source for LinkedIn. The scaffold makes a reasonable assumption: OAuth happens locally, the user stores the refresh token somewhere, the pipeline reads it from there.

LinkedIn Python API Docs | dltHub

Build a LinkedIn-to-database pipeline in Python using dlt with AI Workbench support for Claude Code, Cursor, and Codex.

Request more context

That works fine for a laptop. It breaks in dltHub Runtime, because:

• A headless container can't open a browser to do steps 1-3
• Writing the refresh token to disk doesn't work: the container is destroyed after each run
• Baking the token into the Docker image doesn't work either: tokens expire, and rebuilding the image every rotation is not viable

So the refresh token has to live somewhere that survives across container runs. Some external state store.

Snowflake is the token store

I already have a persisting storage... the pipeline's destination. In this case: Snowflake ❄️ The dltHub container has credentials for it. So, why not also use it to hold the refresh token?

A single-row table in the same schema as the data:

CREATE TABLE IF NOT EXISTS RAW.LINKEDIN_ADS._OAUTH_TOKENS (
    id                       VARCHAR        DEFAULT 'linkedin_ads',
    refresh_token            VARCHAR        NOT NULL,
    refresh_token_expires_at TIMESTAMP_TZ   NOT NULL,
    updated_at               TIMESTAMP_TZ   DEFAULT CURRENT_TIMESTAMP()
);

The pipeline reads the row, exchanges the refresh token for an access token, runs the load, and (if LinkedIn rotated the refresh token) writes the new one back. The table has exactly one row, identified by id = 'linkedin_ads'. Writes use a MERGE so re-running setup or token rotation never duplicates the row.

The setup flow

The only time a human is involved is the initial setup. Once.

I added a --setup flag to the pipeline script. It does the browser dance locally:

1. Build the LinkedIn authorize URL with r_ads_reporting and r_ads scopes
2. Open the browser to that URL
3. Spin up a tiny HTTP server on localhost:8080 to catch the OAuth callback
4. Exchange the code for tokens
5. Persist the refresh token to Snowflake (primary) and to a local JSON file (fallback for local dev)

The redirect URL is configured in my custom app in the LinkedIn Developers portal (this is also where the client id and secret are retrieved from):

The local HTTP server is just Python's http.server. ~20 lines, no Flask, no FastAPI. I overrode log_message to suppress the default access log because nobody needs to see it.

state = secrets.token_urlsafe(16)
auth_url = (
    _AUTH_URL + "?" + urllib.parse.urlencode({
        "response_type": "code",
        "client_id": dlt.secrets["sources.linkedin_ads.client_id"],
        "redirect_uri": _REDIRECT_URI,
        "scope": " ".join(_SCOPES),
        "state": state,
    })
)

server = HTTPServer(("localhost", 8080), _Handler)
server.timeout = 180
webbrowser.open(auth_url)
while not done.is_set():
    server.handle_request()

After --setup succeeds once, every scheduled run is fully automated. No browser, no human, no problem.

Token rotation, handled

The refresh logic on every scheduled run:

token_resp = _refresh_access_token(refresh_token)
access_token = token_resp["access_token"]

new_rt = token_resp.get("refresh_token")
if new_rt and new_rt != refresh_token:
    expires_at = _absolute_expiry(token_resp.get("refresh_token_expires_in", 31536000))
    _write_refresh_token_snowflake(conn, new_rt, expires_at)
    _write_refresh_token_disk(new_rt, expires_at)

Most runs don't write back. LinkedIn only rotates the refresh token occasionally (or when it's close to expiry). The read-only path is the steady state. The write-back path exists so I never have to think about it when rotation does happen.

If LinkedIn ever returns an error on the refresh (status 400, invalid_grant, etc.), the pipeline raises a clear error telling me to re-run --setup. That's the only manual intervention I'll ever need, and it should happen at most once a year if ever 🤞

The rest of the pipeline is standard dlt work: rest_api_resources for accounts/campaigns/creatives, a custom resource for analytics_by_campaign (LinkedIn caps adAnalytics responses at ~118 rows regardless of count). The OAuth plumbing was the only genuinely new part.

And that's it: a scheduled pipeline that authenticates against a 3-legged OAuth API in a container that gets destroyed after every run, by leaning on the same warehouse it's loading data into. Turns out Snowflake makes a pretty good secrets manager when the connection is already open 😎

The full pipeline

"""LinkedIn Ads API to Snowflake pipeline.

dltHub runtime schedule: 02:30 UTC, weekdays (Mon-Fri)
Loads LinkedIn Ads data to RAW.LINKEDIN_ADS schema via 3-legged OAuth.

Auth:  LinkedIn Marketing APIs require 3-legged OAuth. Run setup once locally:
         python linkedin_ads_to_snowflake.py --setup
       The refresh token is stored in RAW.LINKEDIN_ADS._OAUTH_TOKENS in Snowflake
       and refreshed automatically on every run. No token file needs to be baked
       into the dltHub Docker image.

Incremental mode (dltHub job):
  - Loads past 28 days of analytics
  - Merges into existing tables (upsert on date)
  - Runs automatically on schedule

Backfill mode (local only):
  - Loads historical data (all-time, single run)
  - Run locally: python linkedin_ads_to_snowflake.py --backfill
  - Idempotent: re-running merges into existing rows
  - Does NOT run in dltHub jobs

Analytics chunking:
  - LinkedIn's adAnalytics finder caps responses at ~118 rows regardless of
    the `count` param, and the `start` offset is silently ignored. To get the
    full dataset for a wide date range, we split the range into 30-day chunks
    and run one request per chunk.
"""

import json
import os
import secrets
import sys
import urllib.parse
import webbrowser
from datetime import date, datetime, timedelta, timezone
from http.server import BaseHTTPRequestHandler, HTTPServer
from pathlib import Path
from threading import Event
from urllib.parse import urlencode

import dlt
from dlt.sources.rest_api import rest_api_resources
from dlt.sources.helpers.rest_client.client import RESTClient
from dlt.sources.helpers.rest_client.auth import BearerTokenAuth
from dlt.sources.helpers.rest_client.paginators import (
    JSONResponseCursorPaginator,
    SinglePagePaginator,
)
from requests import Request


INCREMENTAL_LOAD_PAST_DAYS = 28
BACKFILL_PAST_DAYS = 365 * 3  # 3 years
ANALYTICS_CHUNK_DAYS = 30  # LinkedIn caps adAnalytics responses at ~118 rows; chunk by date to avoid losing data
LINKEDIN_API_VERSION = "202604"

_AUTH_URL = "https://www.linkedin.com/oauth/v2/authorization"
_TOKEN_URL = "https://www.linkedin.com/oauth/v2/accessToken"
_REDIRECT_URI = "http://localhost:8080/callback"
_SCOPES = ["r_ads_reporting", "r_ads"]
_TOKEN_TABLE = "RAW.LINKEDIN_ADS._OAUTH_TOKENS"
_TOKEN_ROW_ID = "linkedin_ads"


# ---------------------------------------------------------------------------
# Snowflake connection (reuses destination credentials)
# ---------------------------------------------------------------------------

def _snowflake_conn():
    """Open a raw snowflake.connector connection using the dlt destination credentials."""
    from cryptography.hazmat.primitives import serialization
    import snowflake.connector

    creds = dlt.secrets["destination.snowflake.credentials"]
    pem_key = serialization.load_pem_private_key(
        creds["private_key"].encode(),
        password=creds["private_key_passphrase"].encode(),
    )
    pkb = pem_key.private_bytes(
        encoding=serialization.Encoding.DER,
        format=serialization.PrivateFormat.PKCS8,
        encryption_algorithm=serialization.NoEncryption(),
    )
    return snowflake.connector.connect(
        account=creds["host"],
        user=creds["username"],
        private_key=pkb,
        database=creds["database"],
        warehouse=creds["warehouse"],
        role=creds["role"],
    )


# ---------------------------------------------------------------------------
# Token storage: Snowflake (primary) + disk (local dev fallback)
# ---------------------------------------------------------------------------

def _tokens_path() -> Path:
    override = os.environ.get("LINKEDIN_ADS_TOKENS_PATH")
    if override:
        return Path(override)
    return Path(__file__).parent / ".dlt" / "linkedin_ads_tokens.json"


def _read_refresh_token_disk() -> tuple[str, str] | None:
    """Read refresh_token + expires_at from local JSON cache. Returns None if absent."""
    p = _tokens_path()
    if not p.exists():
        return None
    data = json.loads(p.read_text(encoding="utf-8"))
    rt = data.get("refresh_token")
    exp = data.get("refresh_token_expires_at")
    if not rt:
        return None
    return rt, exp or ""


def _write_refresh_token_disk(refresh_token: str, expires_at: str) -> None:
    p = _tokens_path()
    existing = {}
    if p.exists():
        try:
            existing = json.loads(p.read_text(encoding="utf-8"))
        except Exception:
            pass
    existing["refresh_token"] = refresh_token
    existing["refresh_token_expires_at"] = expires_at
    p.parent.mkdir(parents=True, exist_ok=True)
    p.write_text(json.dumps(existing, indent=2), encoding="utf-8")
    try:
        os.chmod(p, 0o600)
    except OSError:
        pass


def _ensure_token_table(conn) -> None:
    """Create the OAuth token schema and table if they don't exist."""
    cur = conn.cursor()
    # Create schema if needed
    cur.execute("CREATE SCHEMA IF NOT EXISTS RAW.LINKEDIN_ADS")
    # Create table
    cur.execute(f"""
        CREATE TABLE IF NOT EXISTS {_TOKEN_TABLE} (
            id                       VARCHAR        DEFAULT '{_TOKEN_ROW_ID}',
            refresh_token            VARCHAR        NOT NULL,
            refresh_token_expires_at TIMESTAMP_TZ   NOT NULL,
            updated_at               TIMESTAMP_TZ   DEFAULT CURRENT_TIMESTAMP()
        )
    """)


def _read_refresh_token_snowflake(conn) -> tuple[str, str] | None:
    """Read refresh_token + expires_at from Snowflake. Returns None if no row."""
    try:
        cur = conn.cursor()
        cur.execute(
            f"SELECT refresh_token, refresh_token_expires_at "
            f"FROM {_TOKEN_TABLE} WHERE id = %s",
            (_TOKEN_ROW_ID,),
        )
        row = cur.fetchone()
        if not row:
            return None
        return str(row[0]), str(row[1])
    except Exception:
        return None


def _write_refresh_token_snowflake(conn, refresh_token: str, expires_at: str) -> None:
    """Upsert refresh_token into Snowflake token table."""
    _ensure_token_table(conn)
    conn.cursor().execute(f"""
        MERGE INTO {_TOKEN_TABLE} t
        USING (SELECT %s AS id) s ON t.id = s.id
        WHEN MATCHED THEN UPDATE SET
            refresh_token            = %s,
            refresh_token_expires_at = %s::TIMESTAMP_TZ,
            updated_at               = CURRENT_TIMESTAMP()
        WHEN NOT MATCHED THEN INSERT (id, refresh_token, refresh_token_expires_at)
            VALUES (%s, %s, %s::TIMESTAMP_TZ)
    """, (
        _TOKEN_ROW_ID,
        refresh_token, expires_at,
        _TOKEN_ROW_ID, refresh_token, expires_at,
    ))


# ---------------------------------------------------------------------------
# LinkedIn OAuth helpers
# ---------------------------------------------------------------------------

def _absolute_expiry(expires_in: int) -> str:
    return (datetime.now(timezone.utc) + timedelta(seconds=int(expires_in))).isoformat()


def _exchange_code(code: str) -> dict:
    """Exchange authorization code for tokens."""
    import requests

    resp = requests.post(
        _TOKEN_URL,
        data={
            "grant_type": "authorization_code",
            "code": code,
            "redirect_uri": _REDIRECT_URI,
            "client_id": dlt.secrets["sources.linkedin_ads.client_id"],
            "client_secret": dlt.secrets["sources.linkedin_ads.client_secret"],
        },
        timeout=30,
    )
    resp.raise_for_status()
    return resp.json()


def _refresh_access_token(refresh_token: str) -> dict:
    """Exchange refresh_token for a new access_token (and possibly new refresh_token)."""
    import requests

    resp = requests.post(
        _TOKEN_URL,
        data={
            "grant_type": "refresh_token",
            "refresh_token": refresh_token,
            "client_id": dlt.secrets["sources.linkedin_ads.client_id"],
            "client_secret": dlt.secrets["sources.linkedin_ads.client_secret"],
        },
        timeout=30,
    )
    if resp.status_code != 200:
        raise RuntimeError(
            f"LinkedIn token refresh failed: {resp.status_code} {resp.text}. "
            "Refresh token may be expired or revoked — re-run: "
            "python linkedin_ads_to_snowflake.py --setup"
        )
    return resp.json()


def _get_access_token() -> str:
    """Return a fresh LinkedIn access token.

    Read order:
      1. Snowflake token table (primary — works in ephemeral dltHub containers)
      2. Local disk cache (fallback for local dev / first run after --setup)

    Always calls the LinkedIn token endpoint to exchange the refresh_token for a
    new access_token. If LinkedIn rotates the refresh_token, writes the new one
    back to Snowflake (and disk).
    """
    conn = _snowflake_conn()
    try:
        result = _read_refresh_token_snowflake(conn)
        if result is None:
            result = _read_refresh_token_disk()
        if result is None:
            raise RuntimeError(
                "No LinkedIn refresh token found in Snowflake or on disk. "
                "Run setup first: python linkedin_ads_to_snowflake.py --setup"
            )
        refresh_token, _ = result

        token_resp = _refresh_access_token(refresh_token)
        access_token = token_resp["access_token"]

        # Write back rotated refresh_token if LinkedIn issued a new one
        new_rt = token_resp.get("refresh_token")
        if new_rt and new_rt != refresh_token:
            expires_at = _absolute_expiry(token_resp.get("refresh_token_expires_in", 31536000))
            _write_refresh_token_snowflake(conn, new_rt, expires_at)
            _write_refresh_token_disk(new_rt, expires_at)

        return access_token
    finally:
        conn.close()


def _do_setup() -> None:
    """Interactive 3-legged OAuth setup.

    Opens browser to LinkedIn authorize URL, captures the callback on
    localhost:8080, exchanges the code for tokens, and stores the refresh_token
    in Snowflake (primary) and on disk (local dev fallback).
    """
    state = secrets.token_urlsafe(16)
    auth_url = (
        _AUTH_URL + "?" + urllib.parse.urlencode({
            "response_type": "code",
            "client_id": dlt.secrets["sources.linkedin_ads.client_id"],
            "redirect_uri": _REDIRECT_URI,
            "scope": " ".join(_SCOPES),
            "state": state,
        })
    )

    code_box: dict[str, str] = {}
    error_box: dict[str, str] = {}
    done = Event()

    class _Handler(BaseHTTPRequestHandler):
        def log_message(self, *args, **kwargs):
            pass

        def do_GET(self):
            params = urllib.parse.parse_qs(urllib.parse.urlparse(self.path).query)
            if params.get("state", [None])[0] != state:
                error_box["error"] = "state_mismatch"
                self._respond(400, "State mismatch. Close this window and retry.")
            elif "error" in params:
                error_box["error"] = params["error"][0]
                self._respond(400, f"LinkedIn denied: {params['error'][0]}")
            elif "code" in params:
                code_box["code"] = params["code"][0]
                self._respond(200, "LinkedIn auth complete. You can close this window.")
            else:
                self._respond(400, "Missing code parameter.")
            done.set()

        def _respond(self, status: int, body: str):
            self.send_response(status)
            self.send_header("Content-Type", "text/plain; charset=utf-8")
            self.end_headers()
            self.wfile.write(body.encode())

    server = HTTPServer(("localhost", 8080), _Handler)
    server.timeout = 180

    print(f"Opening browser to LinkedIn authorize URL...\n  {auth_url}\n")
    if not webbrowser.open(auth_url):
        print("(webbrowser.open returned False — open the URL above manually)")
    print("Waiting for callback on http://localhost:8080/callback ...")
    while not done.is_set():
        server.handle_request()
    server.server_close()

    if error_box:
        raise RuntimeError(f"OAuth failed: {error_box}")
    code = code_box.get("code")
    if not code:
        raise RuntimeError("OAuth callback did not return a code.")

    token_resp = _exchange_code(code)
    refresh_token = token_resp.get("refresh_token")
    if not refresh_token:
        raise RuntimeError("LinkedIn did not return a refresh_token. Check app scopes.")

    expires_at = _absolute_expiry(token_resp.get("refresh_token_expires_in", 31536000))

    # Persist to Snowflake (primary)
    conn = _snowflake_conn()
    try:
        _write_refresh_token_snowflake(conn, refresh_token, expires_at)
        print(f"Refresh token stored in Snowflake ({_TOKEN_TABLE}).")
    finally:
        conn.close()

    # Persist to disk (local dev fallback)
    _write_refresh_token_disk(refresh_token, expires_at)
    print(f"Refresh token also cached on disk ({_tokens_path()}).")
    print(f"  refresh_token expires: {expires_at}")


# ---------------------------------------------------------------------------
# LinkedIn API paginators
# ---------------------------------------------------------------------------

def _flatten_analytics_row(row: dict) -> dict | None:
    """Flatten nested dateRange + pivotValues into top-level PK fields.

    LinkedIn returns analytics rows as:
        {"dateRange": {"start": {"year": ..., "month": ..., "day": ...}, ...},
         "pivotValues": ["urn:li:sponsoredCampaign:123"], ...}
    dlt merge requires flat scalar primary keys, so we add:
        "date_range_start": "YYYY-MM-DD"
        "campaign_urn": "urn:li:sponsoredCampaign:123"
    Returns None if row is missing either PK component (to avoid NULL-key duplicates).
    """
    dr_start = (row.get("dateRange") or {}).get("start") or {}
    try:
        date_range_start = (
            f"{int(dr_start['year']):04d}-"
            f"{int(dr_start['month']):02d}-"
            f"{int(dr_start['day']):02d}"
        )
    except (KeyError, TypeError, ValueError):
        return None
    pv = row.get("pivotValues") or []
    if not pv or not pv[0]:
        return None
    row["date_range_start"] = date_range_start
    row["campaign_urn"] = pv[0]
    # Remove nested fields to prevent dlt from creating unnested subtables
    row.pop("dateRange", None)
    row.pop("pivotValues", None)
    return row


def _linkedin_search_paginator() -> dict:
    """Single-page paginator for LinkedIn search endpoints.

    LinkedIn search endpoints (q=search) return all results in a single response
    and do not support pagination. Using offset/limit params causes repeated
    requests with the same results. Return a single-page paginator instead.
    """
    return {
        "type": "single_page",
    }


class LinkedInCreativesPaginator(JSONResponseCursorPaginator):
    """Cursor paginator that keeps LinkedIn's `List()` finder notation verbatim.

    `requests` URL-encodes `List()` to `List%28%29`, which LinkedIn's Rest.li
    parser rejects with `FIELD_INVALID /value :: array type is not backed by a
    DataList`. We rebuild the URL string with the literal parens and let
    `requests` only encode the dynamic parts (`pageToken`, `pageSize`).
    """

    def _build_url(self, request: Request) -> None:
        params = dict(request.params or {})
        token = params.pop("pageToken", None)
        page_size = params.pop("pageSize", 100)
        query_parts = [
            ("q", "criteria"),
            ("intendedStatuses", "List()"),
            ("pageSize", str(page_size)),
        ]
        if token:
            query_parts.append(("pageToken", str(token)))
        base = request.url.split("?", 1)[0]
        request.url = f"{base}?{urlencode(query_parts, safe='()')}"
        request.params = {}

    def init_request(self, request: Request) -> None:
        super().init_request(request)
        self._build_url(request)

    def update_request(self, request: Request) -> None:
        super().update_request(request)
        self._build_url(request)


class LinkedInSinglePagePaginator(SinglePagePaginator):
    """Single-page paginator that rebuilds the URL preserving Rest.li notation.

    LinkedIn's `List(urn:...)`, `dateRange=(start:...,end:...)`, etc. need
    mixed encoding: `dateRange` keeps colons literal, but URN-based params
    (e.g. `accounts`) must URL-encode their inner colons. We rebuild the URL
    from `request.params` with per-param `safe` chars to keep each Rest.li
    syntax intact without breaking URNs.
    """

    PER_PARAM_SAFE = {
        "dateRange": "():,",
        "accounts": "()",
    }

    def init_request(self, request: Request) -> None:
        super().init_request(request)
        self._build_url(request)

    def _build_url(self, request: Request) -> None:
        params = dict(request.params or {})
        if not params:
            return
        base = request.url.split("?", 1)[0]
        encoded = "&".join(
            urlencode([(k, str(v))], safe=self.PER_PARAM_SAFE.get(k, ""))
            for k, v in params.items()
        )
        request.url = f"{base}?{encoded}"
        request.params = {}


def _linkedin_date_range(start: date, end: date) -> str:
    """Build a LinkedIn Rest.li date range string.

    Format: `(start:(year:YYYY,month:M,day:D),end:(year:YYYY,month:M,day:D))`
    Colons stay literal; outer parens stay literal; comma becomes `%2C` via
    `urlencode(safe="():,")`. See `LinkedInSinglePagePaginator`.
    """
    return (
        f"(start:(year:{start.year},month:{start.month},day:{start.day}),"
        f"end:(year:{end.year},month:{end.month},day:{end.day}))"
    )


# ---------------------------------------------------------------------------
# dlt resources + source
# ---------------------------------------------------------------------------

@dlt.resource(
    name="analytics_by_campaign",
    write_disposition="merge",
    primary_key=["date_range_start", "campaign_urn"],
    max_table_nesting=0,    
)
def linkedin_analytics_chunks(
    access_token: str,
    account_id: str,
    start_date: date,
    end_date: date,
):
    """Yield LinkedIn adAnalytics rows in 30-day chunks.

    LinkedIn caps adAnalytics responses at ~118 rows regardless of `count`,
    and the `start` offset param is silently ignored. Splitting the date range
    into `ANALYTICS_CHUNK_DAYS` chunks keeps each request under the cap.
    The custom paginator handles LinkedIn's Rest.li URL encoding.
    """
    client = RESTClient(
        base_url="https://api.linkedin.com/rest",
        auth=BearerTokenAuth(token=access_token),
        headers={
            "LinkedIn-Version": LINKEDIN_API_VERSION,
            "X-Restli-Protocol-Version": "2.0.0",
        },
    )
    paginator = LinkedInSinglePagePaginator()

    chunk_start = start_date
    while chunk_start <= end_date:
        chunk_end = min(chunk_start + timedelta(days=ANALYTICS_CHUNK_DAYS - 1), end_date)
        params = {
            "q": "analytics",
            "pivot": "CAMPAIGN",
            "timeGranularity": "DAILY",
            "dateRange": _linkedin_date_range(chunk_start, chunk_end),
            "accounts": f"List(urn:li:sponsoredAccount:{account_id})",
        }
        for row in client.paginate(
            path="adAnalytics",
            method="GET",
            params=params,
            paginator=paginator,
            data_selector="elements",
        ):
            for r in row:
                flattened = _flatten_analytics_row(r)
                if flattened:
                    yield flattened
        chunk_start = chunk_end + timedelta(days=1)


@dlt.source(
        name="linkedin_ads_to_snowflake",
        max_table_nesting=0,
)
def linkedin_ads_source(
    access_token: str,
    account_id: str,
    backfill: bool = False,
):
    """Extract LinkedIn Ads data via API v2 using dlt's rest_api_resources.

    Authenticates via Bearer token (3-legged OAuth). Fetches:
    - Account metadata
    - Campaign details
    - Creative metadata
    - Ad analytics by campaign (incremental or backfill)

    Args:
        access_token: Valid LinkedIn access token (from _get_access_token()).
        account_id: LinkedIn ads account ID.
        backfill: If True, load all-time data. If False, load past 28 days.
    """

    end_dt = date.today()
    start_dt = end_dt - timedelta(
        days=BACKFILL_PAST_DAYS if backfill else INCREMENTAL_LOAD_PAST_DAYS
    )

    config = {
        "client": {
            "base_url": "https://api.linkedin.com/rest",
            "auth": {
                "type": "bearer",
                "token": access_token,
            },
            "headers": {
                "LinkedIn-Version": LINKEDIN_API_VERSION,
                "X-Restli-Protocol-Version": "2.0.0",
                "X-RestLi-Method": "FINDER",
            },
        },
        "resources": [
            {
                "name": "accounts",
                "endpoint": {
                    "path": "adAccounts",
                    "params": {"q": "search"},
                    "data_selector": "elements",
                    "paginator": _linkedin_search_paginator(),
                },
                "write_disposition": "replace",
            },
            {
                "name": "campaigns",
                "endpoint": {
                    "path": f"adAccounts/{account_id}/adCampaigns",
                    "params": {"q": "search"},
                    "data_selector": "elements",
                    "paginator": _linkedin_search_paginator(),
                },
                "write_disposition": "replace",
            },
            {
                "name": "creatives",
                "endpoint": {
                    "path": f"adAccounts/{account_id}/creatives",
                    "params": {
                        "q": "criteria",
                        "intendedStatuses": "List()",
                        "pageSize": 100,
                    },
                    "data_selector": "elements",
                    "paginator": LinkedInCreativesPaginator(
                        cursor_path="metadata.nextPageToken",
                        cursor_param="pageToken",
                    ),
                },
                "write_disposition": "replace",
            },
        ],
    }

    yield from rest_api_resources(config)  # type: ignore[arg-type]
    yield linkedin_analytics_chunks(
        access_token=access_token,
        account_id=account_id,
        start_date=start_dt,
        end_date=end_dt,
    )


# ---------------------------------------------------------------------------
# Pipeline entry points
# ---------------------------------------------------------------------------

def load_linkedin_ads(backfill: bool = False) -> None:
    """Load LinkedIn Ads data to Snowflake.

    Reads the refresh_token from RAW.LINKEDIN_ADS._OAUTH_TOKENS, exchanges it
    for a fresh access_token, and runs the pipeline. If LinkedIn rotates the
    refresh_token, the new one is written back to Snowflake automatically.

    Args:
        backfill: If True, load all-time data (local only). If False, load past 28 days.
    """
    access_token = _get_access_token()
    account_id = dlt.secrets["sources.linkedin_ads.account_id"]

    pipeline = dlt.pipeline(
        pipeline_name="linkedin_ads_to_snowflake",
        destination="snowflake",
        staging=dlt.destinations.filesystem(bucket_url="az://stage/linkedin_ads"),
        dataset_name="linkedin_ads",
    )
    
    source = linkedin_ads_source(
        access_token=access_token,
        account_id=account_id,
        backfill=backfill,
    )

    load_info = pipeline.run(source, loader_file_format="jsonl")
    print(load_info)
    load_info.raise_on_failed_jobs()


def main(args: list[str]) -> None:
    if "--setup" in args:
        _do_setup()
        return
    load_linkedin_ads(backfill="--backfill" in args)


if __name__ == "__main__":
    main(sys.argv[1:])