This lesson covers advanced API calling techniques: handling pagination to fetch all pages of data, managing rate limits to avoid being blocked, refreshing authentication tokens, working with streaming APIs, and building resilient API clients. These skills are essential for integrating with any modern web service.
🕯️ Magic Note
A well-designed API client abstracts away the complexity of HTTP requests, pagination, and error handling. The caller should not need to know that the API uses pagination or that tokens expire. The client handles these details automatically.
Python
import requests
import time
def fetch_all_pages(base_url, params=None, page_param=”page”, per_page=100):
“””Fetch all pages from a paginated API.”””
all_items = []
page = 1
while True:
# Add page parameter
query_params = params.copy() if params else {}
query_params[page_param] = page
query_params[“per_page”] = per_page
response = requests.get(base_url, params=query_params)
response.raise_for_status()
data = response.json()
# Extract items (depends on API structure)
items = data.get(“items”, data.get(“results”, data.get(“data”, [])))
if not items:
break
all_items.extend(items)
# Check if there are more pages
total_pages = data.get(“total_pages”, data.get(“pages”))
if total_pages and page >= total_pages:
break
# Check via next page URL (common in REST APIs)
next_url = data.get(“next”, data.get(“next_page_url”))
if next_url is None:
break
page += 1
time.sleep(0.5) # Be polite
print(f”Fetched {len(all_items)} items from {page} pages”)
return all_items
# Usage with GitHub API (link header pagination)
def fetch_github_repos(username):
“””Fetch all repositories for a GitHub user (handles link header pagination).”””
all_repos = []
url = f”https://api.github.com/users/{username}/repos”
while url:
response = requests.get(url)
response.raise_for_status()
all_repos.extend(response.json())
# GitHub uses Link header for pagination
url = None
if “next” in response.links:
url = response.links[“next”][“url”]
time.sleep(0.5)
print(f”Fetched {len(all_repos)} repositories”)
return all_repos
🕯️ Magic Note
APIs use different pagination styles. Some use page and per_page parameters. Others use a cursor or offset. The GitHub API uses Link headers. Always check the API documentation for the specific pagination method.
Python
import requests
import time
from functools import wraps
class RateLimiter:
“””Simple rate limiter for API calls.”””
def __init__(self, calls_per_second=1):
self.calls_per_second = calls_per_second
self.min_interval = 1.0 / calls_per_second
self.last_call_time = 0
def wait_if_needed(self):
elapsed = time.time() – self.last_call_time
if elapsed < self.min_interval:
time.sleep(self.min_interval – elapsed)
self.last_call_time = time.time()
def retry_with_backoff(max_retries=5, base_delay=1, backoff_factor=2):
“””Decorator that retries failed API calls with exponential backoff.”””
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
delay = base_delay
for attempt in range(max_retries):
try:
return func(*args, **kwargs)
except requests.exceptions.RequestException as e:
if attempt == max_retries – 1:
raise
print(f”Attempt {attempt + 1} failed: {e}. Retrying in {delay}s…”)
time.sleep(delay)
delay *= backoff_factor
return None
return wrapper
return decorator
# Example of handling 429 (Too Many Requests)
def handle_rate_limits(response):
if response.status_code == 429:
retry_after = int(response.headers.get(“Retry-After”, 5))
print(f”Rate limited. Waiting {retry_after} seconds…”)
time.sleep(retry_after)
return True # Should retry
return False
Python
import requests
import time
from datetime import datetime, timedelta
class TokenManager:
“””Manages API tokens with automatic refresh.”””
def __init__(self, client_id, client_secret, token_url):
self.client_id = client_id
self.client_secret = client_secret
self.token_url = token_url
self.access_token = None
self.expires_at = None
def refresh_token(self):
“””Obtain a new access token.”””
data = {
“grant_type”: “client_credentials”,
“client_id”: self.client_id,
“client_secret”: self.client_secret
}
response = requests.post(self.token_url, data=data)
response.raise_for_status()
token_data = response.json()
self.access_token = token_data[“access_token”]
expires_in = token_data.get(“expires_in”, 3600)
self.expires_at = datetime.now() + timedelta(seconds=expires_in)
print(f”Token refreshed. Expires at {self.expires_at}”)
def get_token(self):
“””Get a valid token, refreshing if necessary.”””
if not self.access_token or datetime.now() >= self.expires_at – timedelta(minutes=5):
self.refresh_token()
return self.access_token
class APIClient:
“””API client with automatic token management.”””
def __init__(self, base_url, token_manager):
self.base_url = base_url
self.token_manager = token_manager
self.session = requests.Session()
def request(self, method, endpoint, **kwargs):
token = self.token_manager.get_token()
headers = kwargs.pop(“headers”, {})
headers[“Authorization”] = f”Bearer {token}”
url = f”{self.base_url}/{endpoint.lstrip(‘/’)}”
response = self.session.request(method, url, headers=headers, **kwargs)
if response.status_code == 401:
# Token might be expired, force refresh and retry
self.token_manager.refresh_token()
headers[“Authorization”] = f”Bearer {self.token_manager.get_token()}”
response = self.session.request(method, url, headers=headers, **kwargs)
return response
Python
import requests
import json
def stream_twitter_api(bearer_token, stream_url=”https://api.twitter.com/2/tweets/sample/stream”):
“””Process a streaming API (Twitter sample stream).”””
headers = {“Authorization”: f”Bearer {bearer_token}”}
with requests.get(stream_url, headers=headers, stream=True) as response:
if response.status_code != 200:
print(f”Error: {response.status_code}”)
return
for line in response.iter_lines():
if line:
try:
tweet = json.loads(line.decode(“utf-8”))
# Process each tweet as it arrives
print(f”Tweet ID: {tweet.get(‘data’, {}).get(‘id’, ‘N/A’)}”)
except json.JSONDecodeError:
print(f”Could not parse: {line}”)
🕯️ Magic Note
Use stream=True when downloading large responses or processing streaming APIs. This prevents loading the entire response into memory at once. The iter_lines() method yields one line at a time as it arrives.
Python
import requests
import time
from typing import Dict, List, Any, Optional
from functools import wraps
class ResilientAPIClient:
“””Resilient API client with retries, rate limiting, and pagination.”””
def __init__(self, base_url, max_retries=3, rate_limit_per_second=2):
self.base_url = base_url.rstrip(“/”)
self.session = requests.Session()
self.max_retries = max_retries
self.rate_limit_per_second = rate_limit_per_second
self.min_interval = 1.0 / rate_limit_per_second
self.last_request_time = 0
def _rate_limit(self):
elapsed = time.time() – self.last_request_time
if elapsed < self.min_interval:
time.sleep(self.min_interval – elapsed)
self.last_request_time = time.time()
def _request_with_retry(self, method, endpoint, **kwargs):
self._rate_limit()
url = f”{self.base_url}/{endpoint.lstrip(‘/’)}”
delay = 1
for attempt in range(self.max_retries):
try:
response = self.session.request(method, url, **kwargs)
if response.status_code == 429:
retry_after = int(response.headers.get(“Retry-After”, delay))
print(f”Rate limited. Waiting {retry_after}s…”)
time.sleep(retry_after)
continue
if response.status_code >= 500:
print(f”Server error {response.status_code}. Retrying…”)
time.sleep(delay)
delay *= 2
continue
response.raise_for_status()
return response
except requests.exceptions.RequestException as e:
if attempt == self.max_retries – 1:
raise
print(f”Request failed: {e}. Retrying in {delay}s…”)
time.sleep(delay)
delay *= 2
raise Exception(“Max retries exceeded”)
def get(self, endpoint, params=None):
return self._request_with_retry(“GET”, endpoint, params=params)
def post(self, endpoint, data=None, json=None):
return self._request_with_retry(“POST”, endpoint, data=data, json=json)
def get_paginated(self, endpoint, page_param=”page”, per_page=100, max_pages=None):
“””Fetch all pages from a paginated endpoint.”””
all_items = []
page = 1
while True:
response = self.get(endpoint, params={page_param: page, “per_page”: per_page})
data = response.json()
items = data.get(“items”, data.get(“results”, data.get(“data”, [])))
if not items:
break
all_items.extend(items)
if max_pages and page >= max_pages:
break
# Check if there are more pages
total_pages = data.get(“total_pages”)
if total_pages and page >= total_pages:
break
page += 1
return all_items
Python
import requests
import time
from typing import Dict, List, Optional
class GitHubClient:
“””Client for GitHub REST API.”””
def __init__(self, token: Optional[str] = None):
self.base_url = “https://api.github.com”
self.session = requests.Session()
if token:
self.session.headers.update({“Authorization”: f”Bearer {token}”})
self.session.headers.update({“Accept”: “application/vnd.github.v3+json”})
def _handle_rate_limit(self, response):
if response.status_code == 403 and “X-RateLimit-Remaining” in response.headers:
remaining = int(response.headers.get(“X-RateLimit-Remaining”, 0))
if remaining == 0:
reset_time = int(response.headers.get(“X-RateLimit-Reset”, 0))
wait_time = reset_time – time.time() + 5
if wait_time > 0:
print(f”Rate limit exceeded. Waiting {wait_time:.0f} seconds…”)
time.sleep(wait_time)
return True
return False
def _request(self, method, endpoint, **kwargs):
url = f”{self.base_url}/{endpoint.lstrip(‘/’)}”
while True:
response = self.session.request(method, url, **kwargs)
if self._handle_rate_limit(response):
continue
response.raise_for_status()
break
return response
def get_user(self, username: str) -> Dict:
“””Get user information.”””
response = self._request(“GET”, f”users/{username}”)
return response.json()
def get_repos(self, username: str, per_page: int = 100) -> List[Dict]:
“””Get all repositories for a user (handles pagination).”””
all_repos = []
page = 1
while True:
response = self._request(“GET”, f”users/{username}/repos”, params={“page”: page, “per_page”: per_page})
repos = response.json()
if not repos:
break
all_repos.extend(repos)
if len(repos) < per_page:
break
page += 1
return all_repos
def get_stargazers(self, owner: str, repo: str) -> List[Dict]:
“””Get users who starred a repository.”””
all_stargazers = []
page = 1
per_page = 100
while True:
response = self._request(“GET”, f”repos/{owner}/{repo}/stargazers”, params={“page”: page, “per_page”: per_page})
stargazers = response.json()
if not stargazers:
break
all_stargazers.extend(stargazers)
if len(stargazers) < per_page:
break
page += 1
return all_stargazers
# Usage
# client = GitHubClient(token=”your_github_token”)
# user = client.get_user(“octocat”)
# repos = client.get_repos(“octocat”)
# print(f”User: {user[‘name’]}, Repos: {len(repos)}”)
Python
import pytest
from unittest.mock import Mock, patch
import requests
def test_api_client():
# Mock the requests.get function
mock_response = Mock()
mock_response.status_code = 200
mock_response.json.return_value = {“name”: “Test User”, “id”: 123}
with patch(“requests.get”, return_value=mock_response):
response = requests.get(“https://api.example.com/user”)
assert response.status_code == 200
assert response.json()[“name”] == “Test User”
# Using responses library for more sophisticated mocking
# pip install responses
import responses
@responses.activate
def test_with_responses():
responses.get(
“https://api.example.com/users/1”,
json={“id”: 1, “name”: “Ali”},
status=200
)
response = requests.get(“https://api.example.com/users/1”)
assert response.json()[“name”] == “Ali”
- Not handling pagination (missing data beyond first page)
- Ignoring rate limits (getting blocked)
- Hard-coding API keys (security risk)
- Not setting timeouts (requests can hang indefinitely)
- Not handling network errors (program crashes)
- Assuming all APIs return JSON in the same format
- How do you handle pagination when an API returns a “next” URL in the response?
- Write a function that retries a failed API call with exponential backoff.
- What is the purpose of the stream=True parameter?
- How do you handle a 429 (Too Many Requests) response?
- Write code that automatically refreshes an expired API token.
- How can you test API clients without making real network requests?
⚡ Whisper
APIs are the plumbing of the modern web. Water flows through pipes—data flows through APIs. Your job is to connect the pipes. To fetch, to post, to paginate, to retry. The API may be slow. It may fail. It may limit your rate. Your code must handle all of this. Retry with backoff. Respect rate limits. Refresh expired tokens. Fetch all pages. Build clients that are resilient, not fragile. The difference between a hobby script and production code is error handling. A good API client retries when it fails. It waits when rate limited. It paginates until all data is fetched. It never crashes. It always completes. Build your clients this way. Your users will thank you. Your data will be complete. The API will respect you. Connect the pipes. Let the data flow.
