API-First Integration: Consuming Third-Party APIs in Python

Part 2 of the Python for FDE track. Last updated: October 2026.

Every FDE engagement starts the same way: "here are the API docs, good luck." Your job is to turn a third-party API into reliable data your prototype can use. This post covers the full professional pattern — authentication, pagination, rate-limit handling, error classification, and persisting responses — the way you'd actually write it at a client site.

Setup: a session with auth baked in

Use a requests.Session so headers, auth, and connection pooling are configured once. Most client APIs use an API key header or a bearer token:

import requests

session = requests.Session()
session.headers.update({
    "Authorization": "Bearer " + "YOUR_BEARER_TOKEN",   # or "X-API-Key": "YOUR_API_KEY"
    "User-Agent": "fde-client-integration/1.0",
})

resp = session.get("https://api.client.com/v1/projects", timeout=10)
resp.raise_for_status()
print(resp.json()["data"][0]["name"])
# Acme rollout

OAuth2: the client-credentials flow

Enterprise APIs usually speak OAuth2. The client-credentials flow is the server-to-server variant — no user login, just your service proving who it is to get a short-lived token:

import time

import requests

_token, _token_expiry = None, 0

def get_token(session: requests.Session) -> str:
    """Fetch (and cache until expiry) an OAuth2 access token."""
    global _token, _token_expiry
    if _token and time.time() < _token_expiry - 60:   # reuse with 60s safety margin
        return _token
    r = session.post(
        "https://auth.client.com/oauth/token",
        data={"grant_type": "client_credentials",
              "client_id": "YOUR_CLIENT_ID",
              "client_secret": "YOUR_CLIENT_SECRET"},
        timeout=10,
    )
    r.raise_for_status()
    body = r.json()
    _token = body["access_token"]
    _token_expiry = time.time() + int(body.get("expires_in", 3600))
    return _token

Pagination: get everything, not just page one

Demos built on the first page of results are how you embarrass yourself in front of a client. Cursor pagination is the modern standard — loop until there is no next cursor:

def fetch_all(base_url: str, session: requests.Session) -> list:
    items, cursor = [], None
    while True:
        params = {"limit": 100}
        if cursor:
            params["cursor"] = cursor
        r = session.get(base_url, params=params, timeout=15)
        r.raise_for_status()
        page = r.json()
        items.extend(page["data"])
        cursor = page.get("next_cursor")
        if not cursor:
            return items

projects = fetch_all("https://api.client.com/v1/projects", session)
print(f"Fetched {len(projects)} projects")
# Fetched 342 projects

Rate limits: back off gracefully

Hit a rate limit and the API returns 429. The professional response is exponential backoff — and honor the Retry-After header when the API sends one:

import time

def get_with_retry(session: requests.Session, url: str, tries: int = 5):
    for attempt in range(tries):
        r = session.get(url, timeout=15)
        if r.status_code == 429:
            wait = int(r.headers.get("Retry-After", 2 ** attempt))
            print(f"Rate limited — waiting {wait}s (attempt {attempt + 1})")
            time.sleep(wait)
            continue
        r.raise_for_status()
        return r
    raise RuntimeError(f"Still rate-limited after {tries} tries: {url}")

Error handling: classify before you retry

Not every failure deserves a retry. Classify by status code: retry timeouts and 5xx, refresh tokens on 401, fix your request on 4xx:

import requests

def fetch_project(session: requests.Session, project_id: str) -> dict:
    url = f"https://api.client.com/v1/projects/{project_id}"
    try:
        r = session.get(url, timeout=15)
        r.raise_for_status()
        return r.json()
    except requests.Timeout:
        print("Timed out — safe to retry")
        raise
    except requests.HTTPError as e:
        status = e.response.status_code
        if status in (408, 429) or 500 <= status < 600:
            print(f"Retryable {status} — back off and try again")
        elif status == 401:
            print("Unauthorized — refresh the token, don't retry blindly")
        elif status == 404:
            print("Not found — check the resource id")
        raise

Persist: save raw responses before you transform

Golden FDE rule: save the raw API response first, transform second. When the demo breaks at 9 AM, you want the original data on disk, not a second API call away. JSON for small payloads, NDJSON (one JSON object per line) for large ones:

import json
from pathlib import Path

Path("raw").mkdir(exist_ok=True)

projects = fetch_all("https://api.client.com/v1/projects", session)

Path("raw/projects.json").write_text(json.dumps(projects, indent=2))

with open("raw/projects.ndjson", "w") as f:
    for p in projects:
        f.write(json.dumps(p) + "\n")

print(f"Saved {len(projects)} raw records to raw/")
# Saved 342 raw records to raw/

Key takeaways

• Configure auth once on a requests.Session — API key header, bearer token, or OAuth2 client-credentials with cached expiry.
• Paginate everything — loop on the cursor until it's gone; first-page demos embarrass you.
• On 429, back off exponentially and honor Retry-After.
• Classify errors before retrying: retry 408/429/5xx, refresh tokens on 401, fix the request on other 4xx.
• Save raw responses to disk first (JSON or NDJSON) — transform from the saved copy.

Next in this series: Building Client Prototypes Fast: Streamlit Dashboards in an Afternoon.