Have you ever built a Python CLI tool that works perfectly for a few dozen requests, but then grinds to a halt when you need it to process thousands, or even tens of thousands, of API calls? I certainly have. The frustration of watching a script crawl, limited by synchronous I/O and inefficient connection management, is a common pain point for many developers and data scientists. If you've already built CLI tools and are looking to push their performance boundaries, this post is for you. I'll walk you through how I tackled this challenge, demonstrating a practical approach using asynchronous programming, intelligent connection pooling, and efficient data processing. By the end, you'll have a robust pattern to build CLI tools capable of handling 10,000 requests per minute, turning what once felt like a bottleneck into a blazing-fast operation.
Key Takeaways
- Asynchronous I/O with `asyncio` and `httpx` is crucial for performance when dealing with high-volume network requests, preventing your script from blocking on I/O operations.
- Connection pooling, inherently managed by `httpx.AsyncClient`, significantly reduces overhead by reusing established TCP connections instead of creating new ones for every request.
- Setting a sensible concurrency limit for `asyncio.gather` is vital to balance throughput with API rate limits and local resource constraints.
- Efficient data processing using `pandas` DataFrames provides a fast, memory-efficient way to structure and save large volumes of API response data.
- Robust error handling, especially for network issues and malformed responses, is non-negotiable for production-ready, high-throughput CLI tools.
The Problem
My team often needs to ingest data from various external APIs for our financial analyses and machine learning models. While `requests` is fantastic for single calls, when we scale up to fetching hundreds of thousands of records, often spanning multiple endpoints, a simple loop with `requests.get` becomes agonizingly slow. Each request incurs the overhead of establishing a new TCP connection, performing the HTTP handshake, and then tearing it down. This synchronous, blocking nature means my script spends most of its time waiting for the network, not processing data. The goal was to build a CLI component that could reliably fetch a large volume of data from an API, process it, and save it, all while maintaining a high throughput, ideally hitting 10,000 requests per minute or more.
Data and Sources
For this demonstration, I'm using the JSONPlaceholder Todos API, a free, public API that provides a simple set of mock data. It's an excellent stand-in for real-world APIs when you want to simulate network requests without needing authentication or complex schemas.
Data accessed on 2024-07-29.
Step 1 — Setting Up the Project: Initial Request and Basic Structure
The first step is always to verify basic connectivity and establish a project structure. Before optimizing, I start with a simple, synchronous request to confirm the API is reachable and to understand its response structure. This also lays the groundwork for error handling, which is critical.
The sub-problem here is simply fetching *some* data and seeing what it looks like. We'll use `httpx` for consistency, even in this initial synchronous step, to prepare for the asynchronous shift.
import httpx
import json
BASE_URL = "https://jsonplaceholder.typicode.com"
def fetch_single_todo(todo_id: int) -> dict | None:
"""Fetches a single todo item synchronously."""
try:
response = httpx.get(f"{BASE_URL}/todos/{todo_id}")
response.raise_for_status() # Raise an exception for HTTP errors
return response.json()
except httpx.HTTPStatusError as e:
print(f"HTTP error fetching todo {todo_id}: {e.response.status_code} - {e.response.text}")
return None
except httpx.RequestError as e:
print(f"Network error fetching todo {todo_id}: {e}")
return None
if __name__ == "__main__":
sample_todo = fetch_single_todo(1)
if sample_todo:
print("Successfully fetched sample todo:")
print(json.dumps(sample_todo, indent=2))
This snippet confirms we can hit the API and handle basic HTTP and network errors. The `raise_for_status()` method is a neat `httpx` feature that automatically raises `HTTPStatusError` for 4xx/5xx responses, simplifying error checks.
Step 2 — Implementing Asynchronous Programming
The core limitation of the synchronous approach is that the program waits idly during network latency. To overcome this, we introduce `asyncio` and `httpx`'s asynchronous capabilities. The challenge is to make many requests concurrently without blocking.
I'll define an asynchronous function `fetch_todo` that uses an `httpx.AsyncClient` for making requests. The `AsyncClient` is key for connection pooling, as it maintains a pool of connections that can be reused across requests, drastically reducing overhead. We'll then use `asyncio.gather` to run multiple `fetch_todo` calls concurrently.
import asyncio
import httpx
import json
import time
import pandas as pd
# ... (BASE_URL from previous step) ...
async def fetch_todo(client: httpx.AsyncClient, todo_id: int) -> dict | None:
"""Fetches a single todo item asynchronously."""
try:
response = await client.get(f"{BASE_URL}/todos/{todo_id}")
response.raise_for_status()
return response.json()
except httpx.HTTPStatusError as e:
# print(f"HTTP error fetching todo {todo_id}: {e.response.status_code} - {e.response.text}")
return None
except httpx.RequestError as e:
# print(f"Network error fetching todo {todo_id}: {e}")
return None
except json.JSONDecodeError:
# print(f"JSON decode error for todo {todo_id}")
return None
async def fetch_many_todos(num_todos: int, concurrency_limit: int = 100) -> list[dict]:
"""Fetches multiple todo items concurrently."""
all_todos = []
# httpx.AsyncClient automatically handles connection pooling
async with httpx.AsyncClient(timeout=10.0, limits=httpx.Limits(max_connections=concurrency_limit, max_keepalive_connections=20)) as client:
# Create a list of coroutines
tasks = [fetch_todo(client, i) for i in range(1, num_todos + 1)]
# Process tasks in batches to respect concurrency_limit
for i in range(0, len(tasks), concurrency_limit):
batch = tasks[i:i + concurrency_limit]
results = await asyncio.gather(*batch, return_exceptions=True) # return_exceptions allows individual task failures
for res in results:
if isinstance(res, dict):
all_todos.append(res)
# else: print(f"Task failed: {res}") # Optionally log failed tasks
return all_todos
Here, `fetch_todo` is now an `async` function and uses `await client.get(...)`. The `fetch_many_todos` function creates an `AsyncClient` within an `async with` block, ensuring connections are managed correctly. `asyncio.gather(*tasks)` runs all `fetch_todo` coroutines concurrently. I've also added `return_exceptions=True` to `asyncio.gather` so that if one `fetch_todo` fails, the entire `gather` call doesn't stop, allowing us to collect all successful results. The batching logic helps manage the actual concurrency against the `concurrency_limit`.
Step 3 — Connection Pooling and Data Processing
While `httpx.AsyncClient` inherently provides connection pooling, explicitly configuring `httpx.Limits` within the client can fine-tune its behavior. The `max_connections` parameter directly controls how many concurrent connections the client will maintain, which is crucial for respecting API rate limits and preventing resource exhaustion on your end. The `max_keepalive_connections` helps with connection reuse.
After fetching the data, the next sub-problem is efficiently structuring and storing it. Raw lists of dictionaries are fine, but for any serious analysis or saving to disk, `pandas` DataFrames are superior.
# ... (all previous imports and async functions) ...
async def main_collector(num_requests: int, concurrency: int, output_file: str):
"""Main function to orchestrate data fetching and processing."""
start_time = time.
