Working with APIs — REST, Auth, Pagination, Rate Limits

GET /v1/payments?from=1710633000&to=1710719400&count=100 HTTP/1.1
Host: api.payment-gateway.example.com
Authorization: Bearer sk_live_xxxxxxxxxxxx
Accept: application/json
User-Agent: FreshCart-Pipeline/1.0

HTTP/1.1 200 OK
Content-Type: application/json
X-RateLimit-Limit: 1000
X-RateLimit-Remaining: 847
X-RateLimit-Reset: 1710720000

{
  "count": 100,
  "items": [ ... ],
  "cursor": "eyJpZCI6InBheV94eHh4In0="
}

import os, requests

API_KEY = os.environ['GATEWAY_API_KEY']   # never hardcode

response = requests.get(
    'https://api.payment-gateway.example.com/v1/payments',
    headers={'Authorization': f'Bearer {API_KEY}'},
)

# A custom header (check the vendor's docs for the exact name):
requests.get(url, headers={'X-API-Key': API_KEY})

# Query parameter — avoid when you have a choice: keys end up in
# server access logs and browser history, not just request headers:
requests.get(url, params={'api_key': API_KEY})

import requests, time

def fetch_token(token_url, client_id, client_secret, scope=''):
    response = requests.post(
        token_url,
        data={
            'grant_type': 'client_credentials',
            'client_id': client_id,
            'client_secret': client_secret,
            'scope': scope,
        },
        timeout=30,
    )
    response.raise_for_status()
    return response.json()

{
  "access_token": "eyJhbGciOiJSUzI1NiIs...",
  "expires_in": 3600,
  "token_type": "Bearer"
}

class OAuth2ClientCredentials:
    def __init__(self, token_url, client_id, client_secret, scope=''):
        self.token_url, self.client_id = token_url, client_id
        self.client_secret, self.scope = client_secret, scope
        self._token = None
        self._expires_at = 0

    def get_token(self):
        if self._token and time.time() < self._expires_at - 60:
            return self._token          # cached, with a 60s safety buffer

        data = fetch_token(self.token_url, self.client_id, self.client_secret, self.scope)
        self._token = data['access_token']
        self._expires_at = time.time() + data.get('expires_in', 3600)
        return self._token

    def auth_header(self):
        return {'Authorization': f'Bearer {self.get_token()}'}

def refresh_access_token(refresh_token, client_id, client_secret):
    response = requests.post(
        'https://auth.salesforce.com/services/oauth2/token',
        data={
            'grant_type': 'refresh_token',
            'refresh_token': refresh_token,
            'client_id': client_id,
            'client_secret': client_secret,
        },
    )
    response.raise_for_status()
    return response.json()   # a new access_token (sometimes a new refresh_token too)

import hmac, hashlib, time

def sign_request(method, path, body, secret):
    timestamp = str(int(time.time()))
    string_to_sign = f'{timestamp}\n{method.upper()}\n{path}\n{body}'

    signature = hmac.new(
        secret.encode('utf-8'), string_to_sign.encode('utf-8'), hashlib.sha256,
    ).hexdigest()

    return {'X-Timestamp': timestamp, 'X-Signature': signature}

def verify_webhook_signature(payload_body: bytes, signature_header: str, secret: str) -> bool:
    expected = hmac.new(secret.encode('utf-8'), payload_body, hashlib.sha256).hexdigest()
    return hmac.compare_digest(expected, signature_header)

import base64, json, time

def decode_jwt_payload(token):
    parts = token.split('.')
    payload_b64 = parts[1] + '=' * (4 - len(parts[1]) % 4)   # restore stripped base64 padding
    return json.loads(base64.urlsafe_b64decode(payload_b64))

def is_jwt_expired(token, buffer_seconds=60):
    exp = decode_jwt_payload(token).get('exp')
    return exp is not None and time.time() > (exp - buffer_seconds)

# ?page=3&limit=100  ==  SELECT * FROM payments LIMIT 100 OFFSET 200

def fetch_all_offset(base_url, headers, limit=100):
    page = 1
    while True:
        resp = requests.get(base_url, headers=headers, params={'page': page, 'limit': limit})
        resp.raise_for_status()
        items = resp.json().get('items', [])
        if not items:
            break
        yield from items
        if len(items) < limit:
            break
        page += 1

def fetch_all_cursor(url, headers, params, checkpoint_path=None):
    cursor = load_checkpoint(checkpoint_path)   # None on a fresh start

    while True:
        request_params = {**params, **({'cursor': cursor} if cursor else {})}
        resp = requests.get(url, headers=headers, params=request_params, timeout=30)
        resp.raise_for_status()
        data = resp.json()

        items = data.get('items', [])
        yield from items

        cursor = data.get('cursor')
        save_checkpoint(checkpoint_path, cursor)   # survives a mid-run crash

        if not cursor or not items:
            break

import re

def parse_link_header(link_header):
    if not link_header:
        return {}
    return dict(re.findall(r'<([^>]+)>;\s*rel="([^"]+)"', link_header))

def fetch_all_next_url(start_url, headers):
    url = start_url
    while url:
        resp = requests.get(url, headers=headers, timeout=30)
        resp.raise_for_status()
        data = resp.json()

        yield from data.get('items', [])

        url = data.get('next') or parse_link_header(resp.headers.get('Link')).get('next')

def check_rate_limit_headers(response):
    limit = response.headers.get('X-RateLimit-Limit')
    remaining = response.headers.get('X-RateLimit-Remaining')
    reset = response.headers.get('X-RateLimit-Reset')

    if limit and remaining and int(remaining) < int(limit) * 0.1:
        wait = max(0, int(reset) - int(time.time())) if reset else 5
        print(f'Approaching rate limit — waiting {wait}s for window reset')
        time.sleep(wait + 1)

def handle_rate_limit_response(response):
    retry_after = response.headers.get('Retry-After')
    if retry_after:
        try:
            return float(retry_after)          # most common: seconds
        except ValueError:
            from email.utils import parsedate_to_datetime
            from datetime import datetime, timezone
            retry_dt = parsedate_to_datetime(retry_after)   # rarer: an HTTP date
            return max(0, (retry_dt - datetime.now(timezone.utc)).total_seconds())
    return 5.0   # no header at all — a sane default

import threading

class TokenBucketRateLimiter:
    def __init__(self, calls_per_second, burst_size=None):
        self.rate = calls_per_second
        self.capacity = burst_size or int(calls_per_second)
        self.tokens = float(self.capacity)
        self.last_refill = time.monotonic()
        self._lock = threading.Lock()

    def acquire(self):
        while True:
            with self._lock:
                now = time.monotonic()
                self.tokens = min(self.capacity, self.tokens + (now - self.last_refill) * self.rate)
                self.last_refill = now
                if self.tokens >= 1.0:
                    self.tokens -= 1.0
                    return
            time.sleep(1.0 / self.rate / 2)

if not verify_webhook_signature(body, signature, WEBHOOK_SECRET):
    raise HTTPException(status_code=401)

if event_id in processed_event_ids:
    return {'status': 'already_processed'}   # still 200 — do not reprocess

background_tasks.add_task(process_event, event)
return {'status': 'accepted'}

def process_event(event):
    if event.get('event') == 'payment.captured':
        write_payment_to_db(event['payload']['payment']['entity'])
    elif event.get('event') == 'order.paid':
        update_order_status(event['payload']['order']['entity'])

def reconcile_missed_payments(lookback_hours=2):
    from_ts = int(time.time()) - lookback_hours * 3600
    to_ts = int(time.time())

    new_count = 0
    for payment in fetch_all_cursor(PAYMENTS_URL, auth_header(), {'from': from_ts, 'to': to_ts}):
        if upsert_payment(payment):   # upsert returns True only for a genuinely new row
            new_count += 1

    print(f'Reconciliation: {new_count} payments recovered')

from decimal import Decimal, InvalidOperation

def parse_amount(raw):
    if raw is None:
        return None
    try:
        if isinstance(raw, int):
            return Decimal(raw) / 100        # integer cents → dollars
        return Decimal(str(raw).replace(',', '.'))  # float, string, or European comma
    except InvalidOperation:
        return None

>>> parse_amount(3800)      # Stripe-style integer cents
Decimal('38.00')
>>> parse_amount(38.00)     # a float already in dollars
Decimal('38.00')
>>> parse_amount("38,00")   # a European-formatted string
Decimal('38.00')

from datetime import datetime, timezone

def parse_timestamp(raw):
    if raw is None:
        return None
    if isinstance(raw, (int, float)):
        ts = raw / 1000 if raw > 1e10 else raw   # >1e10 means milliseconds, not seconds
        return datetime.fromtimestamp(ts, tz=timezone.utc)
    if isinstance(raw, str) and ('T' in raw or 'Z' in raw):
        return datetime.fromisoformat(raw.replace('Z', '+00:00'))
    return None

def parse_payment(raw):
    return {
        'payment_id': raw.get('id') or raw.get('payment_id'),
        'amount': parse_amount(raw.get('amount')),
        'currency': raw.get('currency', 'USD'),
        'status': (raw.get('status') or '').lower() or None,
        'created_at': parse_timestamp(raw.get('created_at') or raw.get('created')),
        '_raw': raw,   # keep the original — never silently discard unknown data
    }

def detect_schema_changes(sample, expected_fields):
    seen_fields = {k for record in sample for k in record}
    new_fields = seen_fields - expected_fields
    if new_fields:
        print(f'WARNING: API is returning new fields not in schema: {new_fields}')

import os, json, time, logging, uuid
from pathlib import Path

API_BASE = 'https://api.payment-gateway.example.com/v1'
DLQ_PATH = Path('/data/dlq/payments.ndjson')
RUN_ID = str(uuid.uuid4())

logging.basicConfig(level=logging.INFO)
log = logging.getLogger('payment_ingestion')

limiter = TokenBucketRateLimiter(calls_per_second=8)   # stay under a 10/s API limit

def api_get(path, params, max_retries=5):
    url = f'{API_BASE}{path}'

    for attempt in range(1, max_retries + 1):
        limiter.acquire()
        resp = requests.get(url, headers=auth_header(), params=params, timeout=30)

        if resp.status_code == 200:
            check_rate_limit_headers(resp)
            return resp.json()
        elif resp.status_code == 429:
            time.sleep(handle_rate_limit_response(resp))
        elif resp.status_code in (500, 502, 503, 504):
            time.sleep(min(60, 2 ** attempt))
        else:
            resp.raise_for_status()   # a 4xx — do not retry, something is genuinely wrong

    raise RuntimeError(f'API call failed after {max_retries} attempts')

def fetch_payments(from_ts, to_ts):
    checkpoint_file = Path(f'/data/checkpoints/payments_{from_ts}_{to_ts}.json')
    cursor = json.loads(checkpoint_file.read_text())['cursor'] if checkpoint_file.exists() else None

    while True:
        params = {'from': from_ts, 'to': to_ts, 'count': 100, **({'cursor': cursor} if cursor else {})}
        data = api_get('/payments', params)

        yield from data.get('items', [])

        cursor = data.get('cursor')
        if cursor:
            checkpoint_file.write_text(json.dumps({'cursor': cursor}))
        if not cursor or not data.get('items'):
            break

    checkpoint_file.unlink(missing_ok=True)   # clean up only on a full, successful run

def parse_payment_safe(raw):
    try:
        record = parse_payment(raw)   # from Part 07
        if record['amount'] is None or record['amount'] < 0:
            raise ValueError(f"invalid amount: {raw.get('amount')}")
        return record
    except Exception as e:
        with open(DLQ_PATH, 'a') as f:
            f.write(json.dumps({'error': str(e), 'record': raw}) + '\n')
        log.warning('Record sent to DLQ: %s', e)
        return None

from psycopg2.extras import execute_values

def upsert_batch(records, conn):
    rows = [(r['payment_id'], float(r['amount']), r['currency'], r['status'], r['created_at']) for r in records]
    with conn.cursor() as cur:
        execute_values(cur, """
            INSERT INTO silver.payments (payment_id, amount, currency, status, created_at)
            VALUES %s
            ON CONFLICT (payment_id) DO UPDATE SET status = EXCLUDED.status, updated_at = NOW()
        """, rows)
    conn.commit()
    return len(rows)

def run(run_date):
    log.info('Pipeline started for %s (run_id=%s)', run_date, RUN_ID)
    dt = datetime.strptime(run_date, '%Y-%m-%d').replace(tzinfo=timezone.utc)
    from_ts, to_ts = int(dt.timestamp()), int((dt + timedelta(days=1)).timestamp())

    loaded, skipped, batch = 0, 0, []

    with psycopg2.connect(os.environ['DATABASE_URL']) as conn:
        for raw in fetch_payments(from_ts, to_ts):
            parsed = parse_payment_safe(raw)
            if parsed is None:
                skipped += 1
                continue
            batch.append(parsed)
            if len(batch) >= 5000:
                loaded += upsert_batch(batch, conn)
                batch = []
        if batch:
            loaded += upsert_batch(batch, conn)

    log.info('Pipeline complete: loaded=%d skipped=%d', loaded, skipped)

FreshCart has just signed with a new delivery partner, ShipFast, and you're asked to ingest their daily delivery performance data. The payment pipeline just built is not reusable code here — it's a reusable process, applied to an entirely different API.

Step 1 — read the docs with a DE lens

Auth: API key in a header — simple. Rate limit: 500 requests per minute — comfortable. Pagination: cursor-based — good, the same pattern from Part 04. Webhooks: available, for status changes.

Step 2 — test with curl before writing any code

curl -s -H "X-API-Key: $SHIPFAST_API_KEY" \
     "https://api.shipfast.io/v2/deliveries?date=2026-03-17&limit=5"

{
  "data": [ ... ],
  "pagination": { "cursor": "eyJpZCI6MTI...", "has_more": true, "total": 48234 }
}

One request, and three of Part 04's exact concepts are already confirmed: cursor-based, a total field for validating counts, and a shape close enough to the payment API that the same fetch_all_cursor pattern applies with almost no changes.

Step 3 — identify the data quality risks

The amount field is sometimes an integer, sometimes a float, exactly like Part 07's parse_amount was built to handle. delivered_at is null for undelivered orders. agent_id refers to ShipFast's internal IDs, not FreshCart's — three things the parser handles defensively, reusing Part 07's exact helpers.

Steps 4–6 — build small, backfill, then layer on webhooks

Run for one day first, and compare the API's total against rows actually written — any mismatch means a pagination or parsing bug, caught immediately rather than in production. Backfill 90 days of history with the same checkpointed loop from Part 08's Step 3, so a crash on day 47 resumes from day 47, not day 1. Finally, register a webhook endpoint for status changes, reusing Part 06's four-step handler (verify, dedupe, respond, process) and Part 06's hourly reconciliation job to catch anything missed.

Total time from task assignment to production: about two days. Every step reused a pattern already built for the payment pipeline — only the field names, endpoint URLs, and specific quirks changed.