APIs and JSON for Economists

4.1 Why APIs, and starting the NorthMart API (30 min)

Concept. An API (application programming interface) is a web address that returns data instead of a web page. Statistics agencies, central banks, payment processors and marketing platforms all publish data this way. For causal work, APIs matter because they are where treatment logs live: who got an email, when a price changed, which region a campaign launched in.

For this course, NorthMart has its own API. It serves the same data as data/raw/, but the way a real company system would: behind a key, in pages, with nested fields, and only up to "today".

Step 1. Create the folder api/ and the file api/northmart_api.py:

"""
NorthMart REST API (simulated). Provided for the course: you don't need to edit it.

Start it in its own terminal from the repository root:
    python api/northmart_api.py

It serves the files in data/raw/ as a paginated JSON API at http://127.0.0.1:8000
(interactive docs at http://127.0.0.1:8000/docs).

Like a real company API it:
  - requires an API key in the X-API-Key header
  - returns results in pages (page, page_size, next_page)
  - nests related fields inside objects
  - only shows orders up to the simulated "today" stored in api/clock.txt
    (move the clock with: python api/clock.py 2026-04-26)
"""
from datetime import date
from pathlib import Path

import polars as pl
import uvicorn
from fastapi import Depends, FastAPI, Header, HTTPException, Query

ROOT = Path(__file__).resolve().parent.parent
RAW = ROOT / "data" / "raw"
CLOCK = Path(__file__).resolve().parent / "clock.txt"
API_KEY = "northmart-demo-key"
MAX_PAGE_SIZE = 1000

app = FastAPI(title="NorthMart API", version="1.0")

customers = pl.read_csv(RAW / "customers.csv", try_parse_dates=True)
products = pl.read_csv(RAW / "products.csv")
orders = pl.read_csv(RAW / "orders.csv", try_parse_dates=True)
campaign = pl.read_csv(RAW / "email_campaign.csv", try_parse_dates=True)


def today() -> date:
    """The simulated 'today'. Orders after this date have not happened yet."""
    if CLOCK.exists():
        return date.fromisoformat(CLOCK.read_text().strip())
    return date(2026, 4, 26)


def check_key(x_api_key: str | None = Header(default=None)) -> None:
    if x_api_key != API_KEY:
        raise HTTPException(status_code=401, detail="Missing or invalid X-API-Key header")


def paginate(rows: list[dict], page: int, page_size: int) -> dict:
    start = (page - 1) * page_size
    chunk = rows[start:start + page_size]
    has_more = start + page_size < len(rows)
    return {
        "page": page,
        "page_size": page_size,
        "total": len(rows),
        "next_page": page + 1 if has_more else None,
        "data": chunk,
    }


@app.get("/health")
def health() -> dict:
    return {"status": "ok", "today": today().isoformat()}


@app.get("/customers", dependencies=[Depends(check_key)])
def get_customers(page: int = Query(1, ge=1), page_size: int = Query(500, ge=1, le=MAX_PAGE_SIZE)) -> dict:
    rows = [
        {
            "customer_id": r["customer_id"],
            "signup_date": r["signup_date"].isoformat(),
            "profile": {"region": r["region"], "age": r["age"]},
            "is_loyalty": r["is_loyalty"],
        }
        for r in customers.iter_rows(named=True)
    ]
    return paginate(rows, page, page_size)


@app.get("/products", dependencies=[Depends(check_key)])
def get_products() -> dict:
    return {"data": products.to_dicts()}


@app.get("/orders", dependencies=[Depends(check_key)])
def get_orders(
    page: int = Query(1, ge=1),
    page_size: int = Query(500, ge=1, le=MAX_PAGE_SIZE),
    updated_since: str | None = Query(None, description="ISO timestamp; only orders updated after it"),
) -> dict:
    visible = orders.filter(pl.col("order_date") <= today())
    rows = []
    for r in visible.iter_rows(named=True):
        updated_at = f"{r['order_date'].isoformat()}T23:00:00Z"
        if updated_since and updated_at <= updated_since:
            continue
        rows.append({
            "order_id": r["order_id"],
            "customer_id": r["customer_id"],
            "order_date": r["order_date"].isoformat(),
            "status": r["status"],
            "updated_at": updated_at,
            "line": {
                "product_id": r["product_id"],
                "quantity": r["quantity"],
                "unit_price": r["unit_price"],
                "discount_pct": r["discount_pct"],
            },
        })
    return paginate(rows, page, page_size)


@app.get("/campaigns/{campaign_id}/recipients", dependencies=[Depends(check_key)])
def get_recipients(campaign_id: str) -> dict:
    sent = campaign.filter((pl.col("campaign_id") == campaign_id) & (pl.col("sent_date") <= today()))
    if campaign.filter(pl.col("campaign_id") == campaign_id).height == 0:
        raise HTTPException(status_code=404, detail=f"Unknown campaign {campaign_id}")
    return {
        "campaign_id": campaign_id,
        "data": [{"customer_id": r["customer_id"], "sent_date": r["sent_date"].isoformat()}
                 for r in sent.iter_rows(named=True)],
    }


if __name__ == "__main__":
    uvicorn.run(app, host="127.0.0.1", port=8000)

Step 2. Create api/clock.py:

"""
Move the NorthMart API's simulated 'today'.

    python api/clock.py 2026-02-28    # only the 8 pre-email weeks exist
    python api/clock.py 2026-04-26    # all 16 weeks exist (the default)

The API reads the clock on every request, so no restart is needed.
"""
import sys
from datetime import date
from pathlib import Path

CLOCK = Path(__file__).resolve().parent / "clock.txt"

if len(sys.argv) != 2:
    print(__doc__)
    sys.exit(1)

new_day = date.fromisoformat(sys.argv[1])          # fails loudly on a bad date
CLOCK.write_text(new_day.isoformat())
print(f"NorthMart API 'today' is now {new_day}")

Step 3. Start the API in its own terminal. In VS Code, click + in the Terminal panel to open a second terminal, then:

source .venv/bin/activate
python data/make_northmart.py      # only if data/raw/ is missing
python api/northmart_api.py

INFO:     Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)

Leave this terminal running all day. Do your work in the first terminal. Codespaces will offer to open port 8000 in the browser: add /docs to the address to see interactive documentation of every endpoint.

If you close the Codespace, the API stops. Restart it with python api/northmart_api.py at the start of Days 4 and 5.

4.2 Your first requests: status codes and API keys (45 min)

Concept. Every API call returns a status code and a body:

Create day04/first_request.py:

# day04/first_request.py
import requests

BASE = "http://127.0.0.1:8000"

# 1. A public endpoint: no key needed
r = requests.get(f"{BASE}/health", timeout=10)
print(r.status_code)        # 200 = OK
print(r.json())             # the body, turned into a Python dict

# 2. A protected endpoint without a key
r = requests.get(f"{BASE}/products", timeout=10)
print(r.status_code, r.json())   # 401 = not authorised

# 3. The same request with the key in a header
r = requests.get(f"{BASE}/products", headers={"X-API-Key": "northmart-demo-key"}, timeout=10)
r.raise_for_status()        # stop with an error if the status is not 2xx
products = r.json()["data"]
print(len(products), "products; first one:", products[0])

python day04/first_request.py

200
{'status': 'ok', 'today': '2026-04-26'}
401 {'detail': 'Missing or invalid X-API-Key header'}
40 products; first one: {'product_id': 'P001', 'category': 'Grocery', 'base_price': 23.73}

Notes:

• timeout=10 stops your script from hanging forever if the server never answers. Always set one.
• raise_for_status() turns an error code into a Python error. Without it, a 401 silently gives you an error message where you expected data.
• The key travels in a header, not in the URL, so it doesn't end up in logs and browser history.

Try it. Request the recipients of campaign SPRING26 (endpoint /campaigns/SPRING26/recipients) and count them. Then request a campaign that doesn't exist, WINTER25, and print the status code and message.

import requests

BASE = "http://127.0.0.1:8000"
HEADERS = {"X-API-Key": "northmart-demo-key"}

r = requests.get(f"{BASE}/campaigns/SPRING26/recipients", headers=HEADERS, timeout=10)
r.raise_for_status()
print("Recipients:", len(r.json()["data"]))

r = requests.get(f"{BASE}/campaigns/WINTER25/recipients", headers=HEADERS, timeout=10)
print(r.status_code, r.json()["detail"])

Recipients: 1279
404 Unknown campaign WINTER25

4.3 JSON: dictionaries and lists all the way down (45 min)

Concept. APIs answer in JSON, which maps directly onto Python: JSON objects become dictionaries, JSON arrays become lists. Objects can contain objects, so data is often nested.

Create day04/json_basics.py:

# day04/json_basics.py
import json

import requests

BASE = "http://127.0.0.1:8000"
HEADERS = {"X-API-Key": "northmart-demo-key"}

r = requests.get(f"{BASE}/orders", headers=HEADERS, params={"page_size": 2}, timeout=10)
body = r.json()

# 1. The top level is a dict: some metadata plus a list of records
print(list(body.keys()))
print("total orders:", body["total"], "| next page:", body["next_page"])

# 2. Pretty-print one record to see its shape
first = body["data"][0]
print(json.dumps(first, indent=2))

# 3. Nested fields: a dict inside a dict
print("quantity:", first["line"]["quantity"])
revenue = first["line"]["quantity"] * first["line"]["unit_price"]
print(f"revenue: {revenue:.2f}")

# 4. params= builds the query string for you
print(r.url)

['page', 'page_size', 'total', 'next_page', 'data']
total orders: 18542 | next page: 2
{
  "order_id": "O000001",
  "customer_id": "C00003",
  "order_date": "2026-01-08",
  "status": "completed",
  "updated_at": "2026-01-08T23:00:00Z",
  "line": {
    "product_id": "P011",
    "quantity": 1,
    "unit_price": 61.62,
    "discount_pct": 20
  }
}
quantity: 1
revenue: 61.62
http://127.0.0.1:8000/orders?page_size=2

Two new fields compared with the CSV:

• line groups the product fields in a nested object.
• updated_at records when the API last changed the record. Tomorrow, dlt uses it to load only new data.

Try it. Fetch 5 orders and use a list comprehension to compute each order's revenue; print the list and the total.

import requests

r = requests.get("http://127.0.0.1:8000/orders", headers={"X-API-Key": "northmart-demo-key"},
                 params={"page_size": 5}, timeout=10)
orders = r.json()["data"]
revenues = [o["line"]["quantity"] * o["line"]["unit_price"] for o in orders]
print([round(x, 2) for x in revenues])
print("Total:", round(sum(revenues), 2))

[61.62, 75.36, 183.47, 103.62, 72.6]
Total: 496.67

4.4 Pagination: getting every page (50 min)

Concept. APIs rarely return everything at once. NorthMart returns at most 1,000 records per page and tells you the next_page, or null (Python None) on the last page. Forgetting pagination is one of the most common silent errors in applied work: your sample is the first 500 customers and nobody notices.

Create day04/paginate.py:

# day04/paginate.py
import requests

BASE = "http://127.0.0.1:8000"
HEADERS = {"X-API-Key": "northmart-demo-key"}


def fetch_all(endpoint: str, page_size: int = 1000) -> list[dict]:
    """Follow next_page until the API says there is no more."""
    rows: list[dict] = []
    page = 1
    while page is not None:
        r = requests.get(f"{BASE}/{endpoint}", headers=HEADERS,
                         params={"page": page, "page_size": page_size}, timeout=30)
        r.raise_for_status()
        body = r.json()
        rows.extend(body["data"])
        print(f"  {endpoint}: page {page} -> {len(body['data'])} rows")
        page = body["next_page"]
    return rows


if __name__ == "__main__":
    orders = fetch_all("orders")
    print("Orders fetched:", len(orders))

python day04/paginate.py

  orders: page 1 -> 1000 rows
  ...
  orders: page 19 -> 542 rows
Orders fetched: 18542

while page is not None keeps asking until the API says it's done. The function is generic: it works for any paginated endpoint, which we use straight away.

Try it. Use fetch_all to download all customers, turn them into a Polars DataFrame, and count loyalty members and customers per region. (Hint: put your script in day04/ and use from paginate import fetch_all.)

import polars as pl

from paginate import fetch_all

customers = pl.DataFrame(fetch_all("customers")).unnest("profile")
print(customers.head(3))
print("Loyalty members:", customers["is_loyalty"].sum())
print(customers.group_by("region").len().sort("len", descending=True).head(6))

Loyalty members: 1194
│ ON     ┆ 1401 │
│ QC     ┆ 854  │
...
│ null   ┆ 74   │

4.5 Flattening nested JSON with Polars (35 min)

Concept. Analysis needs flat tables. Polars stores a nested object as a Struct column, and .unnest() spreads its fields into ordinary columns.

Create day04/flatten.py:

# day04/flatten.py
import polars as pl

from paginate import fetch_all

orders_json = fetch_all("orders")

# 1. Polars turns a list of dicts into a DataFrame; nested dicts become a Struct column
orders = pl.DataFrame(orders_json)
print(orders.schema)

# 2. unnest() spreads the struct's fields into ordinary columns
flat = orders.unnest("line").with_columns(pl.col("order_date").str.to_date())
print(flat.head(3))

# 3. Reconcile with the CSV we used on Days 1-3
csv = pl.read_csv("data/raw/orders.csv", try_parse_dates=True)
print("API rows:", flat.height, "| CSV rows:", csv.height)
same = flat.select(csv.columns).sort("order_id").equals(csv.sort("order_id"))
print("Identical to the CSV:", same)

Schema({'order_id': String, ..., 'line': Struct({'product_id': String, 'quantity': Int64, 'unit_price': Float64, 'discount_pct': Int64})})
...
API rows: 18542 | CSV rows: 18542
Identical to the CSV: True

That last line is a reconciliation: the same data obtained two ways must match. Whenever you change a data source, reconcile before you estimate anything.

4.6 Day 4 mini-project: a reusable API client (50 min)

Goal. Write day04/api_client.py, a small client that:

1. Reads the API key from an environment variable, never from the code.
2. Uses one requests.Session with automatic retries on 429 and 5xx errors.
3. Yields pages with a generator (yield), so large endpoints don't have to fit in memory at once.
4. Saves all four endpoints as Parquet in data/api/.

# day04/api_client.py
"""Mini-project: a small, reusable client for the NorthMart API."""
import os
from pathlib import Path

import polars as pl
import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

BASE = os.environ.get("NORTHMART_API_URL", "http://127.0.0.1:8000")
OUT = Path("data/api")


def make_session() -> requests.Session:
    """One session for all calls: shared headers, automatic retries on server errors."""
    key = os.environ.get("NORTHMART_API_KEY")
    if not key:
        raise RuntimeError("Set NORTHMART_API_KEY first, e.g.  export NORTHMART_API_KEY=northmart-demo-key")
    session = requests.Session()
    session.headers["X-API-Key"] = key
    retry = Retry(total=3, backoff_factor=1, status_forcelist=[429, 500, 502, 503, 504])
    session.mount("http://", HTTPAdapter(max_retries=retry))
    session.mount("https://", HTTPAdapter(max_retries=retry))
    return session


def fetch_pages(session: requests.Session, endpoint: str, **params):
    """Yield one page of records at a time (a generator)."""
    page = 1
    while page is not None:
        r = session.get(f"{BASE}/{endpoint}", params={"page": page, "page_size": 1000, **params}, timeout=30)
        r.raise_for_status()
        body = r.json()
        yield body["data"]
        page = body.get("next_page")


def fetch_all(session: requests.Session, endpoint: str, **params) -> list[dict]:
    return [row for page in fetch_pages(session, endpoint, **params) for row in page]


def main() -> None:
    OUT.mkdir(parents=True, exist_ok=True)
    s = make_session()

    customers = pl.DataFrame(fetch_all(s, "customers")).unnest("profile")
    orders = pl.DataFrame(fetch_all(s, "orders")).unnest("line")
    products = pl.DataFrame(s.get(f"{BASE}/products", timeout=30).json()["data"])
    recipients = pl.DataFrame(s.get(f"{BASE}/campaigns/SPRING26/recipients", timeout=30).json()["data"])

    for name, df in [("customers", customers), ("orders", orders),
                     ("products", products), ("recipients", recipients)]:
        df.write_parquet(OUT / f"{name}.parquet")
        print(f"{name:<11} {df.height:>7,} rows  -> {OUT / name}.parquet")


if __name__ == "__main__":
    main()

Set the key for this terminal session, then run:

export NORTHMART_API_KEY=northmart-demo-key
python day04/api_client.py

customers     4,000 rows  -> data/api/customers.parquet
orders       18,542 rows  -> data/api/orders.parquet
products         40 rows  -> data/api/products.parquet
recipients    1,279 rows  -> data/api/recipients.parquet

Without the variable you get a clear message instead of a confusing 401:

RuntimeError: Set NORTHMART_API_KEY first, e.g.  export NORTHMART_API_KEY=northmart-demo-key

Why environment variables? Code is shared on GitHub; keys must not be. A key in a script is leaked the moment you push. The demo key is harmless, but build the habit now.

A generator function (yield instead of return) hands back one page, pauses, and continues when asked for the next. Tomorrow you'll see that dlt is built around exactly this idea.

4.7 Save your work with Git (15 min)

git add .
git status          # data/api/ and api/clock.txt should NOT appear
git commit -m "Day 4: NorthMart API, pagination and a reusable client"
git push

Day 4 checklist

• I can start the NorthMart API and read its /docs page
• I can make requests with headers, parameters and a timeout, and read status codes
• I can navigate nested JSON and flatten it with Polars unnest
• I can fetch every page of a paginated endpoint
• I keep API keys in environment variables, not in code