When I first tried to pull user profiles from the Random User API at scale, a single synchronous request loop stalled under a few hundred hits. The API throttles clients at 1 000 requests per minute, and a naïve retry strategy would simply flood the server, leading to 429 errors and a broken pipeline. In this post, I’ll show you how to build a production‑ready CLI that pulls 10 000 requests per minute, respects the API’s limits, and recovers gracefully from transient failures—all while keeping the codebase lean and testable.
Key Takeaways
- Asyncio + aiohttp can turn a 1 000‑RPM bottleneck into a 10 000‑RPM pipeline with the right concurrency controls.
- Token‑bucket throttling provides a clean, time‑based rate limit that scales with concurrency.
- An async retry decorator with exponential backoff and jitter turns flaky HTTP responses into robust data pulls.
The Problem
Data pipelines that rely on external APIs often hit two hard limits: network latency and the provider’s rate cap. A typical synchronous implementation will saturate the CPU with blocking I/O, underutilize the network, and produce a cascade of 429 responses when the provider enforces per‑minute quotas. The result is a slow, unreliable ingestion process that stalls downstream analytics.
Data and Sources
Random User API: https://randomuser.me/api/ – returns a JSON payload with user details. Data accessed on 2026-10-16.
Loading the Data
The first step is a synchronous baseline that demonstrates the throughput ceiling when using requests. It also serves as a sanity check for the API’s response structure.
import requests
def sync_fetch(url):
response = requests.get(url)
response.raise_for_status()
return response.json()
The Core Logic
Below are the building blocks that together enable high‑throughput, resilient ingestion:
1. Async HTTP Client
We use aiohttp.ClientSession to perform non‑blocking GET requests. The session is shared across workers to reuse TCP connections.
import aiohttp
async def fetch(session, url):
async with session.get(url, timeout=10) as resp:
resp.raise_for_status()
return await resp.json()
2. Token‑Bucket Throttler
To respect the 1 000 RPM limit while still allowing 10 000 requests per minute overall, we bucket tokens per second. Each worker must acquire a token before sending a request.
import asyncio
import time
class TokenBucket:
def __init__(self, rate_per_sec):
self._rate = rate_per_sec
self._tokens = rate_per_sec
self._last = time.monotonic()
self._lock = asyncio.Lock()
async def acquire(self):
async with self._lock:
now = time.monotonic()
elapsed = now - self._last
self._tokens += elapsed * self._rate
if self._tokens > self._rate:
self._tokens = self._rate
self._last = now
if self._tokens < 1:
sleep_for = (1 - self._tokens) / self._rate
await asyncio.sleep(sleep_for)
self._tokens = 0
else:
self._tokens -= 1
3. Exponential Backoff with Jitter
A lightweight decorator that retries on transient HTTP errors (5xx, 429) with exponential backoff and random jitter to avoid thundering herd.
import random
def retry(max_attempts=5, base_delay=0.5, backoff=2):
def decorator(func):
async def wrapper(*args, **kwargs):
delay = base_delay
for attempt in range(1, max_attempts + 1):
try:
return await func(*args, **kwargs)
except (aiohttp.ClientError, asyncio.TimeoutError) as e:
if attempt == max_attempts:
raise
jitter = random.uniform(0, delay)
await asyncio.sleep(jitter)
delay *= backoff
return wrapper
return decorator
4. Worker Coroutine
Each worker pulls a token, fetches a URL, and records the result. Errors are logged and counted for diagnostics.
@retry()
async def worker(session, bucket, url, results, errors):
await bucket.acquire()
try:
data = await fetch(session, url)
results.append(data)
except Exception as e:
errors.append(str(e))
5. Orchestrator
The orchestrator creates a pool of workers and runs them concurrently until the target request count is reached.
async def orchestrate(total_requests, concurrency, rate_per_sec):
urls = [f"https://randomuser.me/api/?seed={i}" for i in range(total_requests)]
bucket = TokenBucket(rate_per_sec)
results = []
errors = []
async with aiohttp.ClientSession() as session:
tasks = [
worker(session, bucket, urls[i], results, errors)
for i in range(total_requests)
]
await asyncio.gather(*tasks)
return results, errors
Putting It Together
We expose the orchestrator as a click‑based CLI. Users can tweak concurrency, rate, and output file. The CLI handles argument parsing, validation, and graceful shutdown.
import click
@click.command()
@click.option("--total", default=10000, help="Total requests to send.")
@click.option("--concurrency", default=200, help="Number of concurrent workers.")
@click.option("--rate", default=166.7, help="Tokens per second (1 000 RPM / 6).")
@click.option("--output", default="output.json", help="File to write successful responses.")
def main(total, concurrency, rate, output):
results, errors = asyncio.run(
orchestrate(total, concurrency, rate)
)
with open(output, "w") as f:
json.dump(results, f, indent=2)
click.echo(f"Fetched {len(results)} users, {len(errors)} failures.")
Complete Script
The full runnable script combining all steps:
#!/usr/bin/env python3
import asyncio
import aiohttp
import random
import time
import json
import click
# ---------- Async HTTP Client ----------
async def fetch(session, url):
async with session.get(url, timeout=10) as resp:
resp.raise_for_status()
return await resp.json()
# ---------- Token‑Bucket Throttler ----------
class TokenBucket:
def __init__(self, rate_per_sec):
self._rate = rate_per_sec
self._tokens = rate_per_sec
self._last = time.monotonic()
self._lock = asyncio.Lock()
async def acquire(self):
async with self._lock:
now = time.monotonic()
elapsed = now - self._last
self._tokens += elapsed * self._rate
if self._tokens > self._rate:
self._tokens = self._rate
self._last = now
if self._tokens < 1:
sleep_for = (1 - self._tokens) / self._rate
await asyncio.sleep(sleep_for)
self._tokens = 0
else:
self._tokens -= 1
# ---------- Exponential Backoff ----------
def retry(max_attempts=5, base_delay=0.5, backoff=2):
def decorator(func):
async def wrapper(*args, **kwargs):
delay = base_delay
for attempt in range(1, max_attempts + 1):
try:
return await func(*args, **kwargs)
except (aiohttp.ClientError, asyncio.TimeoutError) as e:
if attempt == max_attempts:
raise
jitter = random.uniform(0, delay)
await asyncio.sleep(jitter)
delay *= backoff
return wrapper
return decorator
# ---------- Worker ----------
@retry()
async def worker(session, bucket, url, results, errors):
await bucket.acquire()
try:
data = await fetch(session, url)
results.append(data)
except Exception as e:
errors.append(str(e))
# ---------- Orchestrator ----------
async def orchestrate(total_requests, concurrency, rate_per_sec):
urls = [f"https://randomuser.me/api/?seed={i}" for i in range(total_requests)]
bucket = TokenBucket(rate_per_sec)
results = []
errors = []
async with aiohttp.ClientSession() as session:
tasks = [
worker(session, bucket, urls[i], results, errors)
for i in range(total_requests)
]
await asyncio.gather(*tasks)
return results, errors
# ---------- CLI ----------
@click.command()
@click.option("--total", default=10000, help="Total requests to send.")
@click.option("--concurrency", default=200, help="Number of concurrent workers.")
@click.option("--rate", default=166.7, help="Tokens per second (1 000 RPM / 6).")
@click.option("--output", default="output.json", help="File to write successful responses.")
def main(total, concurrency, rate, output):
results, errors = asyncio.run(
orchestrate(total, concurrency, rate)
)
with open(output, "w") as f:
json.dump(results, f, indent=2)
click.echo(f"Fetched {len(results)} users, {len(errors)} failures.")
if __name__ == "__main__":
main()
Expected Output
Running python script.py --total 1000 will produce a JSON file with 1 000 user objects. The CLI will print something like:
Fetched 1000 users, 0 failures.
Limitations and Tradeoffs
• The token‑bucket implementation is coarse‑grained; it may under‑utilize bandwidth during bursty traffic.
• Exponential backoff with jitter is simple but does not honor Retry‑After headers from the API.
• The script assumes the API supports a seed parameter for deterministic results; real services may require authentication or pagination.
• High concurrency (200+ workers) can exhaust system resources on low‑end machines; tune --concurrency accordingly.
Frequently Asked Questions
Why not use a library like httpx or tenacity?
Using minimal dependencies keeps the tool lightweight and easier to audit. aiohttp is mature and fully async; tenacity is overkill for simple backoff logic.
How do I respect the API’s Retry‑After header?
Extend the retry decorator to inspect resp.headers.get("Retry-After") and sleep accordingly before the next attempt.
Can this approach handle authentication tokens that expire?
Yes, inject a token refresh routine into the fetch coroutine, ensuring the session always holds a fresh bearer token.
What if I need to write results to a database instead of JSON?
Replace the result‑appending logic in worker with an async database writer (e.g., asyncpg for PostgreSQL).
What I'd Change
In a real deployment I’d replace the manual token bucket with a proven library like aiolimiter to avoid subtle race conditions. I’d also surface Retry‑After handling and integrate a circuit‑breaker pattern to shut down the worker pool during sustained failures. Finally, I’d expose metrics via Prometheus to monitor request rates, success ratios, and backoff durations, turning the CLI into a self‑observing ingestion service.