Extract — Reading from APIs, Files, and Databases

The first stage is the hardest to get right

Most pipeline failures live in the extract stage, because that's where your code touches systems you don't control. APIs throttle you, return malformed JSON on Tuesdays, and change response schemas without telling you. CSV exports change format after a vendor upgrade. Databases drift their column types as ORMs evolve. The discipline of extract is to capture the source-of-truth bytes faithfully, log what you got, and let validation fight the type wars in a separate stage.

The two patterns

Bulk full extract — pull the entire source every run. Simple, idempotent, no state. Fine when the source is small or rarely changes.
Incremental extract — pull only what's changed since the last run, using a watermark column (updated_at >= last_run) or a change feed. Faster, but you have to manage the watermark.

The non-negotiables

Retries with exponential backoff on transient errors (HTTP 429/5xx).
Pagination handled correctly — most APIs return ≤1000 rows per page; loop until the last page returns less than a full page or the API tells you it's done.
Capture the raw response, not just the parsed result. If the API returns junk on Tuesday, you want the raw JSON to debug from.
Set a User-Agent identifying your pipeline. When the upstream operator emails about traffic spikes, they should be able to find you.

Code

A defensive paginated API extractor with retries·python

import time
import logging
from typing import Iterator
import httpx

log = logging.getLogger('extract')

def fetch_orders(client: httpx.Client, since: str) -> Iterator[dict]:
    url = 'https://api.example.com/v1/orders'
    params = {'updated_since': since, 'page_size': 1000}
    page = 1
    while True:
        for attempt in range(5):
            try:
                resp = client.get(url, params={**params, 'page': page})
                if resp.status_code in (429, 500, 502, 503, 504):
                    raise httpx.HTTPStatusError('transient', request=resp.request, response=resp)
                resp.raise_for_status()
                break
            except httpx.HTTPStatusError as e:
                wait = 2 ** attempt
                log.warning('transient error %s, retry in %ds', e.response.status_code, wait)
                time.sleep(wait)
        else:
            raise RuntimeError(f'page {page} exhausted retries')

        body = resp.json()
        for row in body['results']:
            yield row

        if not body.get('next_page'):
            return
        page += 1

with httpx.Client(
    headers={'User-Agent': 'cwk-de-pipeline/1.0 (ops@example.com)'},
    timeout=30.0,
) as client:
    rows = list(fetch_orders(client, since='2026-04-01T00:00:00Z'))

Exercise

Pick any public REST API (GitHub, OpenWeather, a stock ticker free tier). Write a paginated extractor that handles at least one transient-error retry and writes the raw responses to disk before parsing them. Then write a separate function that reads the raw files and parses them into a DataFrame. Notice how the separation makes both halves independently testable.