Python Working with APIs and HTTP Requests: requests Library, REST APIs, Authentication, Pagination, Error Handling, Rate Limiting, and Every Pattern Data Engineers Need

Half the data in your pipeline does not come from databases. It comes from APIs — Salesforce sends customer data as JSON, Stripe sends payment events via webhooks, weather services expose forecasts through REST endpoints, and your company’s microservices communicate exclusively via HTTP. If you cannot call an API, parse the response, handle pagination, manage authentication, and deal with errors — you cannot build a real data pipeline.

Python’s requests library makes HTTP calls simple. But “simple” is only the first 5%. The other 95% — pagination, retry, rate limiting, authentication, error handling, streaming large responses, and timeouts — is what separates a script from a production pipeline.

Real-life analogy: Calling an API is like ordering from a restaurant. You (the client) send a request to the kitchen (the server): “I would like the chicken (GET /chicken).” The kitchen prepares the dish and sends it back with a status: “200 OK — here is your chicken” or “404 Not Found — we do not serve that.” Sometimes you need to prove you are a paying customer (authentication). Sometimes the kitchen is busy and asks you to wait (rate limiting). Sometimes your order is too large for one plate, so they bring it in courses (pagination). And sometimes the kitchen catches fire (500 Internal Server Error) — you need a plan for that too.

Table of Contents

• HTTP Basics for Data Engineers
• What is an API?
• HTTP Methods
• Status Codes
• Headers and Content Types
• The requests Library
• Installing requests
• GET — Retrieving Data
• POST — Sending Data
• PUT and PATCH — Updating Data
• DELETE — Removing Data
• Working with JSON
• Reading JSON Responses
• Sending JSON in Requests
• Nested JSON Navigation
• Query Parameters
• Headers and Authentication
• Custom Headers
• API Key Authentication
• Bearer Token (OAuth)
• Basic Authentication
• Error Handling
• Status Code Checking
• raise_for_status()
• Timeouts
• Connection Errors
• Complete Error Handling Pattern
• Pagination
• Offset-Based Pagination
• Cursor-Based Pagination
• Link Header Pagination
• Complete Paginated Fetch
• Rate Limiting
• Respecting Rate Limits
• Retry-After Headers
• Sessions — Reusing Connections
• Why Use Sessions
• Session Configuration
• Data Engineering Patterns
• Complete API Extractor Class
• Webhook Receiver Pattern
• API Response to DataFrame
• Multi-Endpoint Pipeline
• Incremental API Extraction
• Common Mistakes
• Interview Questions
• Wrapping Up

HTTP Basics for Data Engineers

What is an API?

An API (Application Programming Interface) is a contract between two systems. It defines what requests you can make, what data you send, and what responses you get back. A REST API uses HTTP (the same protocol as web browsers) and exchanges data as JSON. When your pipeline calls GET https://api.example.com/customers, you are making an HTTP request to a server that returns customer data as a JSON response.

HTTP Methods

Method	Purpose	Data Engineering Use	Restaurant Analogy
GET	Retrieve data	Extract records from a source API	Reading the menu
POST	Create new data	Send data to a destination API, trigger a pipeline	Placing an order
PUT	Replace existing data	Full update of a record	Replacing your entire order
PATCH	Partially update data	Update specific fields of a record	Changing just the drink
DELETE	Remove data	Delete old records, cleanup	Canceling an order

Status Codes

Code	Meaning	What Your Pipeline Should Do
200	OK — success	Parse the response and process data
201	Created — resource created	Confirm the POST succeeded
204	No Content — success but no body	Confirm DELETE succeeded (no data to parse)
400	Bad Request — your fault	Fix your request (wrong params, bad JSON)
401	Unauthorized — auth failed	Refresh token or check API key
403	Forbidden — no permission	Check API permissions, contact the API owner
404	Not Found — endpoint or resource missing	Check URL, the resource may have been deleted
429	Too Many Requests — rate limited	Wait and retry (check Retry-After header)
500	Internal Server Error — their fault	Retry with backoff (the server is broken)
502/503	Bad Gateway / Service Unavailable	Retry with backoff (the server is overloaded)

The simple rule: 2xx = success, 4xx = your fault (fix your request), 5xx = their fault (retry).

Headers and Content Types

# Headers are metadata sent with every HTTP request and response
# Common headers:
# Content-Type: tells the server what format you are sending (application/json)
# Accept: tells the server what format you want back (application/json)
# Authorization: your credentials (API key, bearer token)
# User-Agent: identifies your client (optional but polite)

# Content-Type values:
# application/json — JSON data (most APIs)
# text/csv — CSV data
# application/x-www-form-urlencoded — form data
# multipart/form-data — file uploads

The requests Library

Installing requests

# pip install requests
# requests is not in the standard library — must be installed
# It is the most downloaded Python package (100M+ downloads/month)

GET — Retrieving Data

import requests

# Basic GET request
response = requests.get("https://api.example.com/customers")

# The response object contains everything
print(response.status_code)     # 200
print(response.headers)         # {'Content-Type': 'application/json', ...}
print(response.text)            # Raw response body as string
print(response.json())          # Parsed JSON as Python dict/list
print(response.elapsed)         # How long the request took
print(response.url)             # Final URL (after redirects)

# GET with query parameters
response = requests.get(
    "https://api.example.com/customers",
    params={
        "status": "active",
        "limit": 100,
        "offset": 0,
        "sort": "created_at",
        "order": "desc"
    }
)
# URL becomes: https://api.example.com/customers?status=active&limit=100&offset=0&sort=created_at&order=desc
print(response.url)  # Confirm the full URL

POST — Sending Data

# POST sends data TO the server (create a new record, trigger an action)

# Send JSON data
response = requests.post(
    "https://api.example.com/customers",
    json={                          # json= auto-sets Content-Type: application/json
        "name": "Naveen Vuppula",
        "email": "naveen@example.com",
        "department": "Engineering"
    }
)
print(response.status_code)   # 201 Created
print(response.json())        # {"id": 12345, "name": "Naveen Vuppula", ...}

# Send form data
response = requests.post(
    "https://api.example.com/login",
    data={                          # data= sends as form-encoded
        "username": "admin",
        "password": "secret"
    }
)

# Upload a file
with open("data.csv", "rb") as f:
    response = requests.post(
        "https://api.example.com/upload",
        files={"file": ("data.csv", f, "text/csv")}
    )

PUT and PATCH — Updating Data

# PUT replaces the ENTIRE resource
response = requests.put(
    "https://api.example.com/customers/12345",
    json={"name": "Naveen V", "email": "naveen@new.com", "department": "Data"}
)

# PATCH updates ONLY the specified fields
response = requests.patch(
    "https://api.example.com/customers/12345",
    json={"department": "Data Engineering"}   # Only update department
)

DELETE — Removing Data

response = requests.delete("https://api.example.com/customers/12345")
print(response.status_code)   # 204 No Content (success, no response body)

Working with JSON

Reading JSON Responses

response = requests.get("https://api.example.com/customers")
data = response.json()   # Parses JSON string into Python dict/list

# Typical API response structure
# {
#     "data": [
#         {"id": 1, "name": "Alice", "email": "alice@example.com"},
#         {"id": 2, "name": "Bob", "email": "bob@example.com"}
#     ],
#     "meta": {
#         "total": 1500,
#         "page": 1,
#         "per_page": 100,
#         "total_pages": 15
#     }
# }

customers = data["data"]              # The actual records
total = data["meta"]["total"]         # Total records available
print(f"Page 1: {len(customers)} of {total} customers")

Nested JSON Navigation

# APIs often return deeply nested JSON — flatten it for your pipeline

record = {
    "id": 12345,
    "attributes": {
        "name": "Naveen",
        "contact": {
            "email": "naveen@example.com",
            "phone": {"country": "+1", "number": "4165550123"}
        }
    },
    "relationships": {
        "company": {"data": {"id": 99, "name": "Capgemini"}}
    }
}

# Safe nested access with .get() chains
email = record.get("attributes", {}).get("contact", {}).get("email", "N/A")
company = record.get("relationships", {}).get("company", {}).get("data", {}).get("name", "Unknown")

# Flatten for your data lake
flat_record = {
    "id": record["id"],
    "name": record["attributes"]["name"],
    "email": email,
    "phone": record["attributes"]["contact"]["phone"]["number"],
    "company_name": company,
}
print(flat_record)
# {'id': 12345, 'name': 'Naveen', 'email': 'naveen@example.com', 'phone': '4165550123', 'company_name': 'Capgemini'}

Sending JSON in Requests

When sending data to an API, you have two main options: json= and data=. Understanding the difference prevents hours of debugging “400 Bad Request” errors.

# json= — sends a Python dict as JSON (sets Content-Type: application/json automatically)
# This is what most modern APIs expect
response = requests.post(
    "https://api.example.com/customers",
    json={"name": "Naveen", "email": "naveen@example.com"}
)
# Internally: converts dict to JSON string, sets Content-Type header

# data= — sends as form-encoded (Content-Type: application/x-www-form-urlencoded)
# Used for login forms, OAuth token requests, legacy APIs
response = requests.post(
    "https://api.example.com/login",
    data={"username": "admin", "password": "secret"}
)

# COMMON MISTAKE: using data= with a dict when the API expects JSON
# ❌ This sends form-encoded data, not JSON — API returns 400 Bad Request
response = requests.post(url, data={"name": "Naveen"})

# ✅ This sends JSON — API accepts it
response = requests.post(url, json={"name": "Naveen"})

# If you need to send a pre-built JSON string:
import json
response = requests.post(
    url,
    data=json.dumps({"name": "Naveen"}),
    headers={"Content-Type": "application/json"}
)

Query Parameters

# Use params= instead of manually building URL strings

# ❌ BAD — manual string building (fragile, no URL encoding)
url = f"https://api.example.com/search?q={query}&limit={limit}&offset={offset}"

# ✅ GOOD — params dict (auto-encoded, clean)
response = requests.get(
    "https://api.example.com/search",
    params={
        "q": "data engineer",        # Spaces auto-encoded to %20
        "limit": 50,
        "offset": 100,
        "status": "active",
        "created_after": "2026-01-01",
    }
)
print(response.url)
# https://api.example.com/search?q=data+engineer&limit=50&offset=100&status=active&created_after=2026-01-01

Headers and Authentication

Custom Headers

Headers are metadata key-value pairs sent with every HTTP request. They tell the server who you are, what format you want, and how to authenticate you. You can add any custom header the API requires.

# Send custom headers with any request
response = requests.get(
    "https://api.example.com/data",
    headers={
        "Accept": "application/json",            # Tell server you want JSON back
        "Content-Type": "application/json",      # Tell server you are sending JSON
        "User-Agent": "MyPipeline/1.0",          # Identify your client
        "X-Request-ID": "abc-123-def-456",       # Custom tracking ID for debugging
        "X-API-Version": "2.0",                  # Request a specific API version
    }
)

# Check response headers — useful for debugging and rate limit tracking
print(response.headers["Content-Type"])          # "application/json"
print(response.headers.get("X-RateLimit-Remaining"))  # "47" — requests left
print(response.headers.get("X-RateLimit-Reset"))      # "1783706200" — when limit resets

API Key Authentication

# API keys — the simplest auth method. The key goes in a header or query param.

# In a header (more secure — not visible in URLs/logs)
response = requests.get(
    "https://api.example.com/data",
    headers={"X-API-Key": "your-api-key-here"}
)

# In a query parameter (less secure — visible in URLs and server logs)
response = requests.get(
    "https://api.example.com/data",
    params={"api_key": "your-api-key-here"}
)

# RULE: Never hardcode API keys in your code.
# Use environment variables or a secrets manager.
import os
API_KEY = os.getenv("API_KEY")
if not API_KEY:
    raise ValueError("API_KEY environment variable must be set")

Bearer Token (OAuth)

# Bearer tokens — used by OAuth 2.0 APIs (most modern APIs)
# Step 1: Authenticate with client ID + secret to get a token
# Step 2: Use the token in the Authorization header for all requests

# Step 1: Get the token
auth_response = requests.post(
    "https://auth.example.com/oauth/token",
    data={
        "grant_type": "client_credentials",
        "client_id": os.getenv("CLIENT_ID"),
        "client_secret": os.getenv("CLIENT_SECRET"),
        "scope": "read:customers"
    }
)
token = auth_response.json()["access_token"]
expires_in = auth_response.json()["expires_in"]   # Seconds until expiry

# Step 2: Use the token
response = requests.get(
    "https://api.example.com/customers",
    headers={"Authorization": f"Bearer {token}"}
)

# Tokens expire — build a refresh mechanism
# Check expires_in and re-authenticate before it expires

Basic Authentication

# HTTP Basic Auth — username and password (base64-encoded)
# requests handles the encoding for you

from requests.auth import HTTPBasicAuth

response = requests.get(
    "https://api.example.com/data",
    auth=HTTPBasicAuth("username", "password")
)

# Shorthand (tuple):
response = requests.get(
    "https://api.example.com/data",
    auth=("username", "password")
)

Error Handling

Status Code Checking

The requests library does not raise exceptions for HTTP errors (4xx, 5xx). A 500 Internal Server Error returns a response object just like a 200 OK — your code must check the status code explicitly.

response = requests.get("https://api.example.com/customers")

# Check status code manually
if response.status_code == 200:
    data = response.json()
elif response.status_code == 404:
    print("Resource not found")
elif response.status_code == 401:
    print("Authentication failed — check your API key")
elif response.status_code >= 500:
    print(f"Server error: {response.status_code}")

# Boolean check — True for 2xx, False for 4xx/5xx
if response.ok:     # same as response.status_code < 400
    data = response.json()
else:
    print(f"Request failed: {response.status_code}")

raise_for_status()

Instead of checking status codes manually every time, raise_for_status() raises an HTTPError exception for any 4xx or 5xx response. This is the cleanest approach — let exceptions handle errors instead of if/else chains.

# raise_for_status() — raises HTTPError for bad status codes
response = requests.get("https://api.example.com/customers")
response.raise_for_status()   # Does NOTHING for 200. Raises HTTPError for 4xx/5xx.
data = response.json()        # Only reached if status is 2xx

# Catch the error
try:
    response = requests.get("https://api.example.com/bad-endpoint")
    response.raise_for_status()
except requests.exceptions.HTTPError as e:
    print(f"HTTP Error: {e}")               # "404 Client Error: Not Found"
    print(f"Status: {e.response.status_code}")  # 404
    print(f"Body: {e.response.text[:200]}")     # Error page content

Timeouts

Without a timeout, requests.get(url) waits forever if the server does not respond. In a production pipeline, a hung request blocks all downstream processing indefinitely. Always set a timeout.

# Set a timeout in seconds — request fails if server does not respond in time
response = requests.get("https://api.example.com/data", timeout=30)

# Separate connect and read timeouts: (connect, read)
# Connect timeout: how long to wait for the initial TCP connection
# Read timeout: how long to wait for the server to send data
response = requests.get("https://api.example.com/data", timeout=(5, 30))
# 5 seconds to connect, 30 seconds to read the response

# Handle timeout exceptions
try:
    response = requests.get("https://slow-api.example.com/data", timeout=10)
except requests.exceptions.Timeout:
    print("Request timed out after 10 seconds — server is too slow")
    # Retry, skip, or raise

Real-life analogy: A timeout is like telling a waiter “If my food is not here in 30 minutes, I am leaving.” Without a timeout, you sit at the table forever, waiting for an order that may never come — while your family (downstream pipeline steps) starves at home.

Connection Errors

Connection errors happen when the server is unreachable — DNS resolution fails, the server is down, the network is broken, or a firewall blocks the connection. These are different from HTTP errors (which mean the server responded, just with an error status).

# Handle connection failures
try:
    response = requests.get("https://nonexistent-server.example.com/data", timeout=10)
except requests.exceptions.ConnectionError:
    print("Cannot connect to the server — check URL, DNS, or network")
except requests.exceptions.Timeout:
    print("Server exists but is too slow to respond")
except requests.exceptions.RequestException as e:
    # Catch-all for any requests exception
    print(f"Request failed: {e}")

# In production, combine all error types:
try:
    response = requests.get(url, timeout=30)
    response.raise_for_status()
    data = response.json()
except requests.exceptions.Timeout:
    logger.error("API timeout — retrying")
except requests.exceptions.ConnectionError:
    logger.error("API unreachable — check network")
except requests.exceptions.HTTPError as e:
    logger.error(f"API returned error: {e.response.status_code}")
except requests.exceptions.JSONDecodeError:
    logger.error("API returned non-JSON response — check Content-Type")

Complete Error Handling Pattern

import requests
import time
import logging

logger = logging.getLogger(__name__)

def fetch_with_error_handling(url, params=None, headers=None, 
                               max_retries=3, timeout=30):
    """Production-grade API fetch with comprehensive error handling."""
    
    for attempt in range(1, max_retries + 1):
        try:
            response = requests.get(
                url, params=params, headers=headers, timeout=timeout
            )
            
            # Check status code
            if response.status_code == 200:
                return response.json()
            
            elif response.status_code == 429:
                # Rate limited — respect Retry-After header
                retry_after = int(response.headers.get("Retry-After", 60))
                logger.warning(f"Rate limited. Waiting {retry_after}s...")
                time.sleep(retry_after)
                continue
            
            elif response.status_code >= 500:
                # Server error — retry with backoff
                wait = 2 ** attempt
                logger.warning(f"Server error {response.status_code}. Retrying in {wait}s...")
                time.sleep(wait)
                continue
            
            else:
                # Client error (4xx) — do not retry, raise immediately
                response.raise_for_status()
        
        except requests.exceptions.Timeout:
            logger.warning(f"Timeout on attempt {attempt}/{max_retries}")
            if attempt == max_retries:
                raise
            time.sleep(2 ** attempt)
        
        except requests.exceptions.ConnectionError:
            logger.warning(f"Connection failed on attempt {attempt}/{max_retries}")
            if attempt == max_retries:
                raise
            time.sleep(2 ** attempt)
    
    raise requests.exceptions.RetryError(f"All {max_retries} attempts failed for {url}")

Pagination

Most APIs do not return all records at once. If you have 50,000 customers, the API returns them in pages (100 per page = 500 pages). Your pipeline must iterate through all pages to get the complete dataset.

Real-life analogy: Pagination is like reading a book. The library (API) does not hand you the entire encyclopedia at once — it gives you one volume (page) at a time. You read volume 1, then ask for volume 2, then volume 3, until you have read them all. The three pagination styles are like three library systems: offset-based (“give me volume 3”), cursor-based (“give me the volume after this bookmark”), and link-based (“here is the shelf number for the next volume”).

Offset-Based Pagination

def fetch_all_offset(base_url, headers=None, page_size=100):
    """Fetch all records using offset-based pagination."""
    all_records = []
    offset = 0
    
    while True:
        response = requests.get(
            base_url,
            params={"limit": page_size, "offset": offset},
            headers=headers,
            timeout=30
        )
        response.raise_for_status()
        data = response.json()
        
        records = data.get("data", [])
        all_records.extend(records)
        
        total = data.get("meta", {}).get("total", 0)
        logger.info(f"Fetched {len(all_records)}/{total} records")
        
        if len(all_records) >= total or len(records) == 0:
            break
        
        offset += page_size
    
    return all_records

customers = fetch_all_offset("https://api.example.com/customers")
print(f"Total customers fetched: {len(customers)}")

Cursor-Based Pagination

def fetch_all_cursor(base_url, headers=None, page_size=100):
    """Fetch all records using cursor-based pagination.
    
    The API returns a 'next_cursor' in each response.
    Pass it as a parameter to get the next page.
    When next_cursor is null/empty, you have all the data.
    """
    all_records = []
    cursor = None
    
    while True:
        params = {"limit": page_size}
        if cursor:
            params["cursor"] = cursor
        
        response = requests.get(base_url, params=params, headers=headers, timeout=30)
        response.raise_for_status()
        data = response.json()
        
        records = data.get("data", [])
        all_records.extend(records)
        
        cursor = data.get("meta", {}).get("next_cursor")
        logger.info(f"Fetched {len(all_records)} records (cursor: {cursor})")
        
        if not cursor or len(records) == 0:
            break
    
    return all_records

# Cursor-based is better than offset for:
# - Large datasets (offset becomes slow at high values)
# - Data that changes during pagination (offset can skip/duplicate records)
# - Real-time data streams

Link Header Pagination

def fetch_all_link_header(start_url, headers=None):
    """Fetch all records using Link header pagination (GitHub-style).
    
    The API returns a 'Link' header with URLs for next, prev, first, last.
    Example: <https://api.example.com/items?page=2>; rel="next"
    """
    all_records = []
    url = start_url
    
    while url:
        response = requests.get(url, headers=headers, timeout=30)
        response.raise_for_status()
        
        records = response.json()
        if isinstance(records, dict):
            records = records.get("data", records.get("items", []))
        all_records.extend(records)
        
        # Parse the Link header for the next URL
        url = None
        link_header = response.headers.get("Link", "")
        for part in link_header.split(","):
            if 'rel="next"' in part:
                url = part.split(";")[0].strip().strip("<>")
                break
        
        logger.info(f"Fetched {len(all_records)} records")
    
    return all_records

Complete Paginated Fetch

In production, you need a single function that handles all three pagination styles. This universal paginator auto-detects the style based on the response and handles all the edge cases — empty pages, missing totals, and API inconsistencies.

def fetch_all_auto(base_url, headers=None, page_size=100, max_pages=500):
    """Universal paginated fetch — auto-detects pagination style.
    
    Checks for: cursor, Link header, or offset-based pagination.
    Stops when: no more data, max_pages reached, or empty response.
    """
    all_records = []
    url = base_url
    offset = 0
    page = 0
    
    while url and page < max_pages:
        params = {"limit": page_size, "offset": offset} if url == base_url else {}
        
        response = requests.get(url, params=params, headers=headers, timeout=30)
        response.raise_for_status()
        data = response.json()
        
        # Extract records (APIs use different keys)
        if isinstance(data, list):
            records = data
        else:
            records = data.get("data", data.get("results", data.get("items", [])))
        
        if not records:
            break
        
        all_records.extend(records)
        page += 1
        logger.info(f"Page {page}: fetched {len(all_records)} total records")
        
        # Check for cursor-based pagination
        next_cursor = None
        if isinstance(data, dict):
            meta = data.get("meta", data.get("pagination", {}))
            next_cursor = meta.get("next_cursor", meta.get("cursor", None))
        
        if next_cursor:
            url = base_url
            params = {"limit": page_size, "cursor": next_cursor}
            continue
        
        # Check for Link header pagination
        link_header = response.headers.get("Link", "")
        next_link = None
        for part in link_header.split(","):
            if 'rel="next"' in part:
                next_link = part.split(";")[0].strip().strip("<>")
                break
        
        if next_link:
            url = next_link
            continue
        
        # Fall back to offset-based
        total = data.get("meta", {}).get("total", data.get("total", None))
        if total and len(all_records) >= total:
            break
        
        offset += page_size
        if len(records) < page_size:
            break   # Last page — fewer records than page_size
    
    return all_records

Rate Limiting

Respecting Rate Limits

Rate limiting is the API’s way of saying “slow down.” Most APIs allow a fixed number of requests per time window (e.g., 100 requests/minute). If you exceed the limit, the API returns a 429 Too Many Requests status. Your pipeline must throttle itself to stay under the limit — hammering the API faster just gets you blocked.

import time

def rate_limited_fetch(urls, headers=None, requests_per_second=5):
    """Fetch multiple URLs while respecting rate limits."""
    results = []
    interval = 1.0 / requests_per_second   # Time between requests
    
    for i, url in enumerate(urls):
        start = time.time()
        
        response = requests.get(url, headers=headers, timeout=30)
        
        if response.status_code == 429:
            retry_after = int(response.headers.get("Retry-After", 60))
            logger.warning(f"Rate limited at request {i}. Waiting {retry_after}s...")
            time.sleep(retry_after)
            response = requests.get(url, headers=headers, timeout=30)
        
        response.raise_for_status()
        results.append(response.json())
        
        # Throttle: wait if we are going too fast
        elapsed = time.time() - start
        if elapsed < interval:
            time.sleep(interval - elapsed)
    
    return results

# Process 100 customer detail URLs at 5 requests/second
urls = [f"https://api.example.com/customers/{id}" for id in customer_ids]
details = rate_limited_fetch(urls, headers=auth_headers, requests_per_second=5)

Retry-After Headers

When an API rate-limits you (429 status), it often includes a Retry-After header telling you exactly how long to wait. Always check this header instead of guessing — it is the API telling you when your quota resets.

def fetch_with_retry_after(url, headers=None, max_retries=5):
    """Fetch a URL, respecting Retry-After headers on 429 responses."""
    for attempt in range(1, max_retries + 1):
        response = requests.get(url, headers=headers, timeout=30)
        
        if response.status_code == 200:
            return response.json()
        
        if response.status_code == 429:
            # Retry-After can be seconds (integer) or an HTTP date
            retry_after = response.headers.get("Retry-After", "60")
            
            try:
                wait = int(retry_after)   # Seconds until retry
            except ValueError:
                # HTTP date format: "Wed, 09 Jul 2026 18:30:00 GMT"
                from email.utils import parsedate_to_datetime
                retry_date = parsedate_to_datetime(retry_after)
                wait = max(0, (retry_date - datetime.now(timezone.utc)).total_seconds())
            
            logger.warning(f"Rate limited (429). Retry-After: {wait}s (attempt {attempt})")
            time.sleep(wait)
            continue
        
        response.raise_for_status()
    
    raise Exception(f"Failed after {max_retries} attempts for {url}")

# Also check rate limit headers proactively — BEFORE hitting the limit
# X-RateLimit-Remaining: how many requests you have left
# X-RateLimit-Reset: when the window resets (Unix timestamp)
def check_rate_limit(response):
    remaining = int(response.headers.get("X-RateLimit-Remaining", 999))
    if remaining < 5:
        reset_ts = int(response.headers.get("X-RateLimit-Reset", 0))
        wait = max(0, reset_ts - time.time())
        logger.warning(f"Only {remaining} requests left. Waiting {wait:.0f}s for reset...")
        time.sleep(wait)

Sessions — Reusing Connections

Why Use Sessions

Every call to requests.get() creates a new TCP connection — a fresh handshake with the server. For one request, this is fine. For 500 paginated requests, that is 500 handshakes. A requests.Session reuses the same TCP connection across all requests, persists headers and cookies, and can be 2-5x faster for sequential API calls.

Real-life analogy: Without a session, every API call is like calling a new taxi — you have to give the address, show your ID, and negotiate the fare each time. With a session, you hire a driver for the day — they already know your destination pattern, have your payment on file, and the car is already running. Much faster for multiple trips.

Session Configuration

# A Session reuses the underlying TCP connection — faster for multiple requests
# Also persists headers, cookies, and auth across all requests

# Without session — new connection for every request (slow)
for url in urls:
    response = requests.get(url, headers={"Authorization": f"Bearer {token}"})

# With session — reuses connection, persists headers (fast)
session = requests.Session()
session.headers.update({
    "Authorization": f"Bearer {token}",
    "Accept": "application/json",
    "User-Agent": "MyPipeline/1.0"
})
session.timeout = 30   # Default timeout for all requests

for url in urls:
    response = session.get(url)   # Headers are automatically included

# Session as a context manager (auto-closes)
with requests.Session() as session:
    session.headers.update({"Authorization": f"Bearer {token}"})
    
    customers = session.get("https://api.example.com/customers").json()
    orders = session.get("https://api.example.com/orders").json()

Data Engineering Patterns

Complete API Extractor Class

import requests
import time
import logging
from datetime import datetime, timezone

logger = logging.getLogger(__name__)

class APIExtractor:
    """Production-grade API data extractor with auth, pagination, retry, and rate limiting."""
    
    def __init__(self, base_url, auth_token=None, api_key=None,
                 requests_per_second=10, max_retries=3, timeout=30):
        self.base_url = base_url.rstrip("/")
        self.max_retries = max_retries
        self.timeout = timeout
        self.rate_interval = 1.0 / requests_per_second
        self.last_request_time = 0
        
        self.session = requests.Session()
        self.session.headers.update({"Accept": "application/json"})
        
        if auth_token:
            self.session.headers["Authorization"] = f"Bearer {auth_token}"
        if api_key:
            self.session.headers["X-API-Key"] = api_key
    
    def _throttle(self):
        """Enforce rate limiting between requests."""
        elapsed = time.time() - self.last_request_time
        if elapsed < self.rate_interval:
            time.sleep(self.rate_interval - elapsed)
        self.last_request_time = time.time()
    
    def _request(self, method, endpoint, **kwargs):
        """Make an HTTP request with retry and error handling."""
        url = f"{self.base_url}/{endpoint.lstrip('/')}"
        kwargs.setdefault("timeout", self.timeout)
        
        for attempt in range(1, self.max_retries + 1):
            self._throttle()
            try:
                response = self.session.request(method, url, **kwargs)
                
                if response.status_code == 429:
                    retry_after = int(response.headers.get("Retry-After", 60))
                    logger.warning(f"Rate limited. Waiting {retry_after}s...")
                    time.sleep(retry_after)
                    continue
                
                if response.status_code >= 500:
                    wait = 2 ** attempt
                    logger.warning(f"Server error {response.status_code}. Retry in {wait}s...")
                    time.sleep(wait)
                    continue
                
                response.raise_for_status()
                return response
            
            except (requests.exceptions.Timeout, requests.exceptions.ConnectionError) as e:
                if attempt == self.max_retries:
                    raise
                wait = 2 ** attempt
                logger.warning(f"Attempt {attempt} failed: {e}. Retry in {wait}s...")
                time.sleep(wait)
        
        raise Exception(f"All {self.max_retries} attempts failed for {url}")
    
    def get(self, endpoint, params=None):
        return self._request("GET", endpoint, params=params)
    
    def fetch_all(self, endpoint, params=None, page_size=100, data_key="data"):
        """Fetch all records with automatic pagination."""
        all_records = []
        params = dict(params or {})
        params["limit"] = page_size
        offset = 0
        
        while True:
            params["offset"] = offset
            response = self.get(endpoint, params=params)
            data = response.json()
            
            records = data.get(data_key, [])
            all_records.extend(records)
            
            total = data.get("meta", {}).get("total", len(records))
            logger.info(f"Fetched {len(all_records)}/{total} from {endpoint}")
            
            if len(all_records) >= total or len(records) == 0:
                break
            offset += page_size
        
        return all_records
    
    def close(self):
        self.session.close()
    
    def __enter__(self):
        return self
    
    def __exit__(self, *args):
        self.close()

# Usage
with APIExtractor(
    "https://api.example.com",
    auth_token=os.getenv("API_TOKEN"),
    requests_per_second=5
) as api:
    customers = api.fetch_all("/customers", params={"status": "active"})
    orders = api.fetch_all("/orders", params={"created_after": "2026-07-01"})
    
    print(f"Extracted {len(customers)} customers, {len(orders)} orders")

API Response to DataFrame

import pandas as pd

def api_to_dataframe(records, flatten=True):
    """Convert a list of API records (nested JSON) to a flat DataFrame."""
    if flatten:
        flat_records = [flatten_dict(r) for r in records]
        return pd.DataFrame(flat_records)
    return pd.DataFrame(records)

def flatten_dict(d, parent_key="", sep="_"):
    """Recursively flatten a nested dictionary."""
    items = []
    for k, v in d.items():
        new_key = f"{parent_key}{sep}{k}" if parent_key else k
        if isinstance(v, dict):
            items.extend(flatten_dict(v, new_key, sep).items())
        elif isinstance(v, list):
            items.append((new_key, str(v)))   # Convert lists to string
        else:
            items.append((new_key, v))
    return dict(items)

# Usage
with APIExtractor("https://api.example.com", auth_token=TOKEN) as api:
    records = api.fetch_all("/customers")

df = api_to_dataframe(records)
print(df.head())
print(f"Columns: {list(df.columns)}")
# Nested fields like {"contact": {"email": "a@b.com"}} become "contact_email"

Webhook Receiver Pattern

Webhooks are the reverse of API calls — instead of your pipeline pulling data from an API, the external system pushes data to your endpoint when events happen. Think of it as the API calling you. Common examples: Stripe sends payment events, GitHub sends push events, Salesforce sends record change events.

# A simple webhook receiver using Flask
# In production, use a proper WSGI server (gunicorn) behind a load balancer

from flask import Flask, request, jsonify
import json
import logging
from datetime import datetime, timezone

app = Flask(__name__)
logger = logging.getLogger(__name__)

@app.route("/webhooks/payments", methods=["POST"])
def receive_payment_webhook():
    """Receive payment events from Stripe-like service."""
    
    # Step 1: Verify the webhook signature (prevents spoofing)
    signature = request.headers.get("X-Webhook-Signature")
    if not verify_signature(request.data, signature):
        return jsonify({"error": "Invalid signature"}), 401
    
    # Step 2: Parse the event
    event = request.json
    event_type = event.get("type")
    event_data = event.get("data", {})
    
    logger.info(f"Webhook received: {event_type}")
    
    # Step 3: Write to your data lake (append to a JSON file or queue)
    webhook_record = {
        "received_at": datetime.now(timezone.utc).isoformat(),
        "event_type": event_type,
        "payload": event_data,
    }
    
    with open(f"webhooks/{event_type}_{datetime.now().strftime('%Y%m%d')}.jsonl", "a") as f:
        f.write(json.dumps(webhook_record) + "\n")
    
    # Step 4: Return 200 quickly — process later
    # Webhook senders retry if they don't get 200 within a few seconds
    return jsonify({"status": "received"}), 200

# In production:
# - Use a message queue (SQS, Kafka) instead of writing to files
# - Process webhooks asynchronously (return 200 fast, process in background)
# - Store raw events in bronze layer, transform in silver
# - Implement idempotency (same event received twice should not create duplicates)

Multi-Endpoint Pipeline

Real-world APIs rarely have all data in one endpoint. Customers are in /customers, orders in /orders, products in /products. A complete extraction requires calling multiple endpoints, potentially joining data across them, and handling each one’s pagination and error modes independently.

import logging
from datetime import datetime, timezone

logger = logging.getLogger(__name__)

def run_api_pipeline(api_config):
    """Extract data from multiple API endpoints with per-endpoint settings.
    
    api_config = {
        "customers": {"page_size": 100, "params": {"status": "active"}},
        "orders":    {"page_size": 500, "params": {"created_after": "2026-01-01"}},
        "products":  {"page_size": 200, "params": {}},
    }
    """
    results = {}
    
    with APIExtractor(
        base_url="https://api.example.com",
        auth_token=os.getenv("API_TOKEN"),
        requests_per_second=5
    ) as api:
        for endpoint, config in api_config.items():
            logger.info(f"Extracting: /{endpoint}")
            start = datetime.now(timezone.utc)
            
            try:
                records = api.fetch_all(
                    f"/{endpoint}",
                    params=config.get("params", {}),
                    page_size=config.get("page_size", 100)
                )
                
                elapsed = (datetime.now(timezone.utc) - start).total_seconds()
                results[endpoint] = {
                    "records": records,
                    "count": len(records),
                    "status": "success",
                    "duration": elapsed,
                }
                logger.info(f"  ✅ {endpoint}: {len(records)} records in {elapsed:.1f}s")
                
            except Exception as e:
                elapsed = (datetime.now(timezone.utc) - start).total_seconds()
                results[endpoint] = {
                    "records": [],
                    "count": 0,
                    "status": "failed",
                    "error": str(e),
                    "duration": elapsed,
                }
                logger.error(f"  ❌ {endpoint} FAILED after {elapsed:.1f}s: {e}")
    
    # Summary
    total_records = sum(r["count"] for r in results.values())
    failed = [name for name, r in results.items() if r["status"] == "failed"]
    logger.info(f"Pipeline complete: {total_records} total records, {len(failed)} failures")
    
    if failed:
        logger.warning(f"Failed endpoints: {', '.join(failed)}")
    
    return results

# Usage
pipeline_config = {
    "customers": {"page_size": 100, "params": {"status": "active"}},
    "orders":    {"page_size": 500, "params": {"created_after": "2026-07-01"}},
    "products":  {"page_size": 200, "params": {}},
    "invoices":  {"page_size": 100, "params": {"status": "unpaid"}},
}

results = run_api_pipeline(pipeline_config)

# Save each endpoint's data to the bronze layer
for name, result in results.items():
    if result["status"] == "success":
        save_to_bronze(result["records"], f"bronze/{name}/")

Incremental API Extraction

from datetime import datetime, timezone, timedelta
import json

def incremental_extract(api, endpoint, watermark_file="watermark.json"):
    """Extract only records modified since the last run."""
    
    # Load watermark (last successful extraction timestamp)
    try:
        with open(watermark_file, "r") as f:
            watermark = json.load(f).get(endpoint, "2020-01-01T00:00:00Z")
    except FileNotFoundError:
        watermark = "2020-01-01T00:00:00Z"   # First run — get everything
    
    logger.info(f"Extracting {endpoint} modified since {watermark}")
    
    # Fetch only modified records
    records = api.fetch_all(
        endpoint,
        params={"modified_after": watermark, "sort": "modified_at"}
    )
    
    if records:
        # Update watermark to the latest modified_at
        new_watermark = max(r.get("modified_at", watermark) for r in records)
        
        # Save watermark
        try:
            with open(watermark_file, "r") as f:
                watermarks = json.load(f)
        except FileNotFoundError:
            watermarks = {}
        
        watermarks[endpoint] = new_watermark
        with open(watermark_file, "w") as f:
            json.dump(watermarks, f, indent=2)
        
        logger.info(f"Extracted {len(records)} new/modified records. Watermark: {new_watermark}")
    else:
        logger.info(f"No new records since {watermark}")
    
    return records

# Usage — each run only fetches what changed
with APIExtractor("https://api.example.com", auth_token=TOKEN) as api:
    new_customers = incremental_extract(api, "/customers")
    new_orders = incremental_extract(api, "/orders")

Common Mistakes

1. Not setting timeouts — requests.get(url) without timeout waits forever if the server does not respond. Always set timeout=30 (or appropriate value). A hung request blocks your entire pipeline indefinitely.
2. Not checking status codes — response = requests.get(url) does not raise an error on 4xx or 5xx. A 500 error silently returns an HTML error page that your response.json() call then fails to parse with a confusing JSONDecodeError. Always call response.raise_for_status() or check response.status_code.
3. Hardcoding API keys in source code — API keys in code end up in Git history, logs, and error messages. Use environment variables (os.getenv("API_KEY")) or a secrets manager (AWS Secrets Manager, Azure Key Vault). Never commit credentials.
4. Not handling pagination — the first API call returns 100 customers, you process 100, and declare success. But there are 50,000. Always check for pagination metadata (total, next_cursor, Link header) and loop until all pages are fetched.
5. Retrying on 4xx errors — 4xx means your request is wrong (bad URL, missing parameter, invalid auth). Retrying the same bad request 3 times does not fix it. Only retry on 5xx (server errors) and 429 (rate limiting). Fix 4xx errors in your code.
6. Not using sessions for multiple requests — each requests.get() creates a new TCP connection. For 500 paginated calls, that is 500 handshakes. Use requests.Session() to reuse the connection — significantly faster.
7. Ignoring rate limits — hammering an API with hundreds of concurrent requests gets you blocked (429) or banned. Respect rate limits: add delays between requests, check Retry-After headers, and throttle to the API’s documented limit.

Interview Questions

Q: What is the difference between GET and POST? A: GET retrieves data from the server — it is read-only and should not change anything. POST sends data to the server to create a resource or trigger an action. In data pipelines, GET is used for extraction (fetching records), POST is used for loading (sending data to a destination API), and occasionally for authentication (sending credentials to get a token).

Q: How do you handle API pagination in a data pipeline? A: Check the API documentation for the pagination style: offset-based (increment offset by page_size), cursor-based (pass next_cursor from the response), or link-header (follow the rel="next" URL from the Link header). Loop until there is no next page or the total is reached. Collect all records across pages before processing. Cursor-based is preferred for large or real-time datasets because it is consistent even when data changes during pagination.

Q: What is rate limiting and how do you handle it? A: Rate limiting is when an API restricts the number of requests you can make per time period (e.g., 100 requests/minute). When exceeded, the API returns a 429 status code. Handle it by: (1) Throttling your requests to stay under the limit, (2) Checking the Retry-After header for how long to wait, (3) Using exponential backoff on 429 responses. Never retry immediately — it makes the problem worse.

Q: Why should you always set a timeout on API requests? A: Without a timeout, requests.get(url) waits indefinitely if the server does not respond. A pipeline that hangs for hours blocks downstream processes, wastes compute resources, and provides no error to investigate. Setting timeout=30 ensures the request fails after 30 seconds with a Timeout exception that you can catch, log, and retry.

Q: What is the difference between a requests Session and individual requests? A: Individual calls (requests.get()) create a new TCP connection each time. A Session (requests.Session()) reuses the same TCP connection across multiple requests, persists headers and cookies, and is significantly faster for sequential calls. Use sessions when making multiple requests to the same API — pagination, multi-endpoint extraction, or any loop of API calls.

Q: How do you implement incremental extraction from an API? A: Maintain a watermark — the timestamp of the last successfully extracted record. On each run, pass modified_after={watermark} as a query parameter to fetch only new or updated records. After successful extraction, update the watermark to the latest modified_at value from the response. Store the watermark in a file or database so it persists between pipeline runs.

Wrapping Up

APIs are the connective tissue of modern data pipelines. The requests library makes HTTP calls simple, but production pipelines need more: authentication (API keys, OAuth tokens), pagination (offset, cursor, link-header), error handling (status codes, timeouts, retries), rate limiting (throttling, Retry-After), and sessions (connection reuse). The APIExtractor class in this post combines all of these into a reusable component that handles the infrastructure so your pipeline code can focus on the data.

The key patterns: always set timeouts, always check status codes, always handle pagination, never hardcode credentials, and use sessions for multiple requests. These apply whether you are extracting from Salesforce, Stripe, a weather API, or your company’s internal microservices.

This is the final post in Section 2: Intermediate Python. You now have a complete foundation — from file handling and error management through OOP, modules, dates, regex, decorators, and APIs. Next up: Section 3 takes everything you have learned and applies it to real data engineering workflows — Parquet files, database connections, ETL patterns, testing, and parallel processing.

Previous in this series: Decorators and Context Managers

Next in this series: Section 3 — Python for Data Engineering (coming soon)

---

Naveen Vuppula is a Senior Data Engineering Consultant and app developer based in Ontario, Canada. He writes about Python, SQL, AWS, Azure, and everything data engineering at DriveDataScience.com.