Week 1 · Ingest anything › Day 6

Day 6. APIs and source databases

1. Why this matters

Ad platforms, POS clouds, booking engines, CRMs and payment providers all expose REST APIs. Marketing data is the hardest teacher: numbers for "yesterday" keep changing for days as late conversions are attributed. And sometimes the source is another company database you can read from.

2. A well-behaved API client

import time, requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

def session():
    s = requests.Session()
    retry = Retry(total=5, backoff_factor=1.5, status_forcelist=[429, 500, 502, 503, 504],
                  respect_retry_after_header=True, allowed_methods=["GET"])
    s.mount("https://", HTTPAdapter(max_retries=retry))
    s.headers["Authorization"] = f"Bearer {settings.ads_token}"
    return s

def fetch_day(s, day: str):
    url, params, pages = f"{BASE}/reports/daily", {"date": day, "page": 1}, []
    while True:
        r = s.get(url, params=params, timeout=30); r.raise_for_status()
        body = r.json(); pages.append(body)                 # land every page raw, unchanged
        if not body.get("next_page"):
            return pages
        params["page"] = body["next_page"]

Pagination comes in three shapes: page number, offset/limit, and cursor (next token or link). Cursors are the safest when data changes mid-read.

3. Data that changes after the fact

Ad conversions, hotel bookings (modified, cancelled), POS refunds: yesterday's numbers are not final. The pattern is a look-back window: every run re-pulls the last N days and replaces them.

Choose N from data: measure how much a day's total still changes after 1, 3, 7, 14 days. If day+7 changes are under 0.1% of spend, \(N=7\) is a defensible choice, and the report footer says "last 7 days provisional".

4. When the source is a database

Always read from a replica or a read-only login, during agreed hours, with a query timeout. A heavy SELECT on a POS database at lunchtime is how you meet the operations director.

5. dlt: the same thing in 20 lines

dlt (data load tool) is an open-source Python library that handles pagination, incremental state, schema inference and evolution, and loading into DuckDB, Postgres, Azure SQL (mssql), Snowflake, BigQuery and more.

import dlt
from datetime import date, timedelta

def days_between(start: str, end: date):
    d = date.fromisoformat(start)
    while d <= end:
        yield d.isoformat(); d += timedelta(days=1)

@dlt.resource(write_disposition="merge", primary_key=["date", "campaign_id"])
def ad_stats(cursor=dlt.sources.incremental("date", initial_value="2026-09-01")):
    s = session()
    start = (date.fromisoformat(cursor.last_value) - timedelta(days=7)).isoformat()  # 7-day look-back
    for d in days_between(start, date.today() - timedelta(days=1)):
        for page in fetch_day(s, d):
            yield page["data"]          # dlt infers the schema, keeps state, merges on the key

pipeline = dlt.pipeline("ads", destination="duckdb", dataset_name="raw_ads")
print(pipeline.run(ad_stats()))         # swap destination="mssql" or "postgres" later