Topfolio
Tutorial

REST APIs for Data Analysts in Python: Authentication, Pagination & JSON Normalization

Master REST API data extraction in Python. Learn how to handle Bearer tokens and API keys, loop through offset and cursor pagination, flatten nested JSON with pd.json_normalize(), and build fault-tolerant pipelines with automatic retries.

Anuj SainiAug 24, 202612 min read

In modern data organizations, not all valuable data lives neatly in a Postgres data warehouse or an S3 data lake. Critical business metrics originate in third-party SaaS platforms: Stripe (billing events), HubSpot / Salesforce (CRM leads), Shopify (e-commerce orders), GitHub (engineering velocity), and public market APIs (crypto, weather, macroeconomics).

As a data analyst or analytics engineer, extracting data from REST APIs and transforming deeply nested JSON payloads into clean, analysis-ready Pandas DataFrames is a core prerequisite skill.

In this guide, you will master everything from basic HTTP requests to production-grade, fault-tolerant pagination engines.

Interactive Python Practice

Ready to test your data extraction and transformation skills? Solve real-world Python wrangling questions on Topfolio Practice and explore our full Data Analyst Career Track.

1. HTTP Foundations for Data Analysts

A REST API (Representational State Transfer) allows your Python environment to communicate with remote servers over HTTP.

+-------------------+                           +-------------------+
|                   |  1. HTTP GET /orders?p=1  |                   |
|   Python Script   | ------------------------> |   REST API Server |
|  (Requests Client)|                           |  (Stripe/Shopify) |
|                   | <------------------------ |                   |
|                   |  2. Status: 200 OK + JSON |                   |
+-------------------+                           +-------------------+

Essential HTTP Status Codes

Status CodeMeaningPipeline Action
200 OKRequest succeeded; payload returned.Parse JSON with response.json().
400 Bad RequestInvalid query parameters or syntax error.Log payload error and halt.
401 UnauthorizedMissing or expired API key / token.Refresh credentials or alert operator.
403 ForbiddenAuthenticated, but key lacks permissions.Verify API key permission scopes.
404 Not FoundEndpoint URL does not exist.Validate base URL and resource path.
429 Too Many RequestsHit API rate limit.Pause execution and retry with backoff.
500 / 502 / 503Remote server internal failure.Retry with exponential backoff.

2. Secure Authentication Patterns

Never hardcode credentials or secrets in plain text. Always read API keys from environment variables using os.getenv() or a .env file via python-dotenv.

bash
pip install requests python-dotenv pandas

A. Bearer Token Authentication (OAuth2 / JWT)

The most common standard across modern enterprise APIs (Stripe, OpenAI, Supabase, Google Cloud):

python
import os
import requests
from dotenv import load_dotenv
 
load_dotenv()
API_TOKEN = os.getenv("STRIPE_SECRET_KEY")
 
headers = {
    "Authorization": f"Bearer {API_TOKEN}",
    "Accept": "application/json",
    "User-Agent": "TopfolioAnalyticsPipeline/1.0"
}
 
response = requests.get(
    "https://api.stripe.com/v1/charges",
    headers=headers,
    timeout=(3.05, 27) # (Connect timeout, Read timeout)
)
 
if response.status_code == 200:
    data = response.json()
    print(f"Retrieved {len(data.get('data', []))} records successfully.")

B. Custom Header API Keys

Used by services like CoinMarketCap, AlphaVantage, or custom enterprise gateways:

python
headers = {
    "X-API-KEY": os.getenv("COINMARKETCAP_API_KEY"),
    "Content-Type": "application/json"
}
response = requests.get("https://pro-api.coinmarketcap.com/v1/cryptocurrency/listings/latest", headers=headers)

C. Basic Authentication

Encodes username and password into a Base64 authorization header:

python
from requests.auth import HTTPBasicAuth
 
response = requests.get(
    "https://api.github.com/user",
    auth=HTTPBasicAuth(os.getenv("GITHUB_USERNAME"), os.getenv("GITHUB_PAT"))
)

3. Mastering API Pagination

APIs restrict the number of items returned in a single call (typically 20 to 250 records) to protect server performance. Extracting complete datasets requires automated pagination loops.

Type 1: Page-Based Pagination (page & per_page)

Commonly used by GitHub, Shopify, and CoinGecko:

python
import requests
import pandas as pd
import time
 
all_coins = []
base_url = "https://api.coingecko.com/api/v3/coins/markets"
max_pages = 5
 
for page in range(1, max_pages + 1):
    params = {
        "vs_currency": "usd",
        "order": "market_cap_desc",
        "per_page": 50,
        "page": page,
        "sparkline": "false"
    }
    
    response = requests.get(base_url, params=params, timeout=10)
    
    if response.status_code != 200:
        print(f"Error fetching page {page}: {response.status_code}")
        break
        
    records = response.json()
    if not records:
        print(f"Reached end of data stream at page {page}.")
        break
        
    all_coins.extend(records)
    print(f"Fetched page {page} ({len(records)} records)")
    
    # Polite sleep to avoid triggering rate limiters
    time.sleep(0.5)
 
df_crypto = pd.DataFrame(all_coins)
print(f"Total Extracted DataFrame Rows: {len(df_crypto)}")

Type 2: Offset-Based Pagination (limit & offset)

Common in SQL-backed endpoints (e.g., Supabase, PostgREST, Hasura):

python
def fetch_all_offset(base_url, limit=100):
    all_rows = []
    offset = 0
    
    while True:
        params = {"limit": limit, "offset": offset}
        response = requests.get(base_url, params=params, timeout=10)
        response.raise_for_status()
        
        data = response.json()
        batch = data.get("results", [])
        
        if not batch:
            break
            
        all_rows.extend(batch)
        print(f"Fetched offset {offset} to {offset + len(batch)}")
        
        if len(batch) < limit:
            # Last page reached
            break
            
        offset += limit
        time.sleep(0.2)
        
    return pd.DataFrame(all_rows)

Type 3: Cursor-Based Pagination (cursor / starting_after)

The gold standard for high-volume, real-time datasets (Stripe, Slack, Twitter/X). Each request returns an opaque cursor pointing to the next batch:

python
def fetch_all_cursor(api_url, headers, limit=100):
    all_events = []
    has_more = True
    next_cursor = None
    
    while has_more:
        params = {"limit": limit}
        if next_cursor:
            params["starting_after"] = next_cursor
            
        response = requests.get(api_url, headers=headers, params=params, timeout=10)
        response.raise_for_status()
        
        payload = response.json()
        data = payload.get("data", [])
        
        if not data:
            break
            
        all_events.extend(data)
        has_more = payload.get("has_more", False)
        
        # Cursor is the ID of the last item in the batch
        next_cursor = data[-1]["id"]
        print(f"Batch fetched: {len(data)} items. Has more: {has_more}")
        time.sleep(0.2)
        
    return all_events

4. Flattening Complex JSON with pd.json_normalize()

Real-world API responses are almost never flat. They contain nested dictionaries (e.g., user.address.zipcode) and nested lists of child records (e.g., orders[].items[]).

Standard pd.DataFrame(json_data) places whole sub-dictionaries into individual cells. We use pd.json_normalize() to flatten these hierarchies.

Deeply Nested API Response Example

python
nested_api_response = {
    "status": "success",
    "meta": {
        "page": 1,
        "total_records": 2,
        "environment": "production"
    },
    "users": [
        {
            "id": "usr_101",
            "profile": {
                "first_name": "Alice",
                "last_name": "Chen",
                "email": "alice@example.com"
            },
            "subscription": {
                "plan": "Enterprise",
                "mrr": 499.00,
                "status": "active"
            },
            "teams": ["Data", "Growth"]
        },
        {
            "id": "usr_102",
            "profile": {
                "first_name": "Bob",
                "last_name": "Martinez",
                "email": "bob@example.com"
            },
            "subscription": {
                "plan": "Pro",
                "mrr": 99.00,
                "status": "active"
            },
            "teams": ["Engineering"]
        }
    ]
}

Unpacking with pd.json_normalize()

python
# Flatten the 'users' array while retaining parent metadata
df_users = pd.json_normalize(
    data=nested_api_response,
    record_path=['users'],
    meta=[
        ['meta', 'page'],
        ['meta', 'environment']
    ],
    sep='_'
)
 
display(df_users)

Flattened Output DataFrame

idteamsprofile_first_nameprofile_last_nameprofile_emailsubscription_plansubscription_mrrsubscription_statusmeta_pagemeta_environment
usr_101['Data', 'Growth']AliceChenalice@example.comEnterprise499.0active1production
usr_102['Engineering']BobMartinezbob@example.comPro99.0active1production

5. Enterprise Resilience: HTTPAdapter & Retries

Production pipelines must never crash due to a temporary network blip or a transient 502 Bad Gateway from an overloaded server.

We configure requests.Session() with an HTTPAdapter backed by urllib3.util.retry.Retry:

python
import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
 
def create_resilient_session(
    retries=4,
    backoff_factor=1.5,
    status_forcelist=(429, 500, 502, 503, 504)
) -> requests.Session:
    """
    Creates a requests.Session with built-in exponential backoff retries.
    Backoff formula: {backoff factor} * (2 ** ({retry count} - 1))
    e.g. 1.5s -> 3.0s -> 6.0s -> 12.0s
    """
    session = requests.Session()
    
    retry_strategy = Retry(
        total=retries,
        read=retries,
        connect=retries,
        backoff_factor=backoff_factor,
        status_forcelist=status_forcelist,
        raise_on_status=False
    )
    
    adapter = HTTPAdapter(max_retries=retry_strategy)
    session.mount("https://", adapter)
    session.mount("http://", adapter)
    
    return session

6. End-to-End Production Script: Reusable API Extractor

Here is a clean, object-oriented API extraction class that combines auth management, session pooling, cursor pagination, retry handling, and DataFrame normalization:

python
import os
import time
import requests
import pandas as pd
from typing import Dict, Any, Optional
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
 
class APIDataPipeline:
    def __init__(self, base_url: str, api_token: Optional[str] = None):
        self.base_url = base_url.rstrip('/')
        self.session = requests.Session()
        
        # Configure Retries
        retries = Retry(
            total=4,
            backoff_factor=1.0,
            status_forcelist=[429, 500, 502, 503, 504],
            raise_on_status=False
        )
        adapter = HTTPAdapter(max_retries=retries)
        self.session.mount("https://", adapter)
        self.session.mount("http://", adapter)
        
        # Default Headers
        headers = {
            "Accept": "application/json",
            "User-Agent": "TopfolioDataPipeline/2026.1"
        }
        if api_token:
            headers["Authorization"] = f"Bearer {api_token}"
            
        self.session.headers.update(headers)
 
    def extract_paginated(
        self,
        endpoint: str,
        record_key: str,
        params: Optional[Dict[str, Any]] = None,
        max_records: int = 5000
    ) -> pd.DataFrame:
        """
        Extracts paginated records into a flattened Pandas DataFrame.
        """
        all_records = []
        page = 1
        query_params = params.copy() if params else {}
        query_params["per_page"] = 100
        
        url = f"{self.base_url}/{endpoint.lstrip('/')}"
        
        while len(all_records) < max_records:
            query_params["page"] = page
            
            try:
                response = self.session.get(url, params=query_params, timeout=(3.05, 20))
                
                # Check for rate limiting
                if response.status_code == 429:
                    retry_after = int(response.headers.get("Retry-After", 5))
                    print(f"Rate limited (429). Sleeping for {retry_after}s...")
                    time.sleep(retry_after)
                    continue
                    
                if response.status_code != 200:
                    print(f"Failed with status {response.status_code}: {response.text}")
                    break
                    
                payload = response.json()
                
                # Extract records list
                if isinstance(payload, list):
                    batch = payload
                else:
                    batch = payload.get(record_key, [])
                    
                if not batch:
                    break
                    
                all_records.extend(batch)
                print(f"Page {page} fetched. Total items: {len(all_records)}")
                
                if len(batch) < query_params["per_page"]:
                    break
                    
                page += 1
                time.sleep(0.1) # Gentle throttling
                
            except requests.exceptions.RequestException as err:
                print(f"Network error on page {page}: {err}")
                break
                
        # Normalize into DataFrame
        if not all_records:
            return pd.DataFrame()
            
        return pd.json_normalize(all_records, sep='_')
 
# Example Usage with Open-Meteo Weather API (No Auth required)
if __name__ == "__main__":
    client = APIDataPipeline(base_url="https://api.open-meteo.com/v1")
    
    # Fetch London Hourly Weather
    params = {
        "latitude": 51.5074,
        "longitude": -0.1278,
        "hourly": "temperature_2m,relative_humidity_2m,wind_speed_10m"
    }
    
    res = client.session.get("https://api.open-meteo.com/v1/forecast", params=params)
    data = res.json()
    
    df_weather = pd.DataFrame(data['hourly'])
    df_weather['time'] = pd.to_datetime(df_weather['time'])
    print(f"\nExtracted Weather DataFrame ({len(df_weather)} rows):")
    display(df_weather.head())

Summary & Next Steps

You now possess the foundational engineering tools to ingest data from any REST API in Python:

  1. Construct authenticated requests with Bearer tokens and API headers.
  2. Automate multi-page extractions across Page, Offset, and Cursor pagination paradigms.
  3. Flatten multi-level JSON structures with pd.json_normalize().
  4. Build self-healing pipelines with HTTPAdapter, exponential retry backoffs, and rate-limit awareness.

Explore our other technical data guides to build end-to-end analytics pipelines:

Frequently Asked Questions

How do I choose between Basic Auth, API Keys, and Bearer Tokens in Python?

The target API specifies its auth requirements in its documentation. Basic Auth passes encoded username/password via HTTPBasicAuth. API Keys are passed either as custom headers (e.g., 'X-API-KEY') or query parameters. Bearer tokens (OAuth2/JWT) must be passed in the standard Authorization header as 'Bearer <token>'.

What is the difference between Offset-based and Cursor-based pagination?

Offset-based pagination (?limit=100&offset=200) requests data by row skips, which suffers from performance degradation on deep offsets and data duplication if new records are inserted during extraction. Cursor-based pagination (?cursor=eyJpZCI6...) uses an immutable pointer to the last record seen, providing O(1) database index lookups and immune to pagination drift.

Why does standard pd.DataFrame() fail on nested JSON, and how does pd.json_normalize() solve it?

pd.DataFrame() places nested JSON objects and arrays into single DataFrame cells as raw dictionaries and lists, making SQL-style aggregation impossible. pd.json_normalize() flattens nested key-value hierarchies into distinct dot-separated columns (e.g., user.address.city) and extracts embedded child record arrays into individual rows.

How should Python scripts handle HTTP 429 (Rate Limit Exceeded) and 500 server errors?

Do not crash the pipeline. Use requests.Session() configured with an HTTPAdapter and urllib3.util.retry.Retry specifying status_forcelist=[429, 500, 502, 503, 504], backoff_factor=1 (exponential sleep 1s, 2s, 4s...), and read the 'Retry-After' response header if provided.

Why should you always use requests.Session() instead of raw requests.get()?

requests.Session() reuses underlying TCP connections (HTTP Keep-Alive), saving SSL handshake latency across hundreds of paginated requests. It also maintains shared headers, default timeouts, and auth tokens across all child requests.

Anuj Saini

Written by

Anuj SainiFounder & Lead Instructor

Founder at Topfolio with 6+ years in data & analytics across JPMC, Ultrahuman, and high-growth startups. Sat on hiring panels, reviewed 500+ resumes, and writes practical SQL & data guides.

View all articles

Related Articles

Tutorial

Customer Analytics in Python: Cohort Analysis, RFM Segmentation & LTV

Master customer analytics in Python with Pandas and Seaborn. Calculate RFM scores, build customer segments, run monthly cohort retention heatmaps, and estimate Customer Lifetime Value (LTV).

13 min readRead
Tutorial

Market Basket Analysis in Python: Support, Confidence, Lift & Apriori Explained

Master Market Basket Analysis and Association Rule Mining in Python. Learn the mathematical intuition behind Support, Confidence, and Lift, one-hot encode transaction baskets, and implement Apriori and FP-Growth using Pandas and Mlxtend.

12 min readRead
Tutorial

Python Exploratory Data Analysis (EDA): The Complete Step-by-Step Workflow

Master the complete 6-stage Python Exploratory Data Analysis (EDA) framework. Learn structured data inspection, missing value imputation, IQR outlier detection, distribution analysis, correlation heatmaps, and feature profiling.

13 min readRead