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'}

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

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

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

Rate Limiting

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)

Sessions — Reusing Connections

# 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()

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.

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"

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.