When I first tried to harvest every article from Discord Engineering’s RSS feed, my simple `requests` loop stalled after a few dozen items. I needed a tool that could fetch thousands of links per minute, respect rate limits, and survive transient failures without blowing up. This post shows how to replace the naïve `requests` pattern with an async, rate‑limited, retry‑capable CLI that writes a JSON Lines file in minutes, not hours.
Key Takeaways
- Async I/O with httpx and asyncio lets you launch thousands of concurrent HTTP requests with minimal CPU usage.
- Combining a semaphore, aiolimiter, and dynamic “Retry‑After” handling keeps you within server limits and protects your IP.
- Tenacity’s retry decorator with exponential backoff and jitter makes transient errors disappear.
- A simple circuit‑breaker prevents wasting traffic on endpoints that keep failing.
- Click, logging, and tqdm make the tool usable, observable, and easy to integrate into pipelines.
The Problem
Data‑hungry services often need to pull content from dozens or hundreds of external APIs. A synchronous, blocking approach scales poorly: each request blocks the thread, so you can only issue a handful of requests per second before hitting the I/O wall. Worse, when the upstream API starts throttling or returns intermittent 5xx errors, a naïve loop can hang forever or crash the whole process. I needed a CLI that could ingest 10 000 requests per minute while gracefully handling rate limits, retries, and failures.
Data and Sources
The live data for this exercise comes from Discord Engineering’s RSS feed: https://discord.com/blog/rss.xml. Feed parsing library: feedparser HTTP client: httpx Asynchronous rate limiter: aiolimiter Retry library: tenacity CLI parsing: click HTML parsing: BeautifulSoup4 Data accessed on 2026-10-22.
Loading the Data
First we pull the RSS feed and extract a list of article URLs. We keep only the first fetch_limit links to keep the demo fast.
import feedparser
def load_rss(rss_url, fetch_limit):
feed = feedparser.parse(rss_url)
urls = [entry.link for entry in feed.entries[:fetch_limit]]
return urls
The Core Logic
Below we break the implementation into five sub‑steps that mirror the outline: async foundation, rate limiting, retries, circuit breaking, and production readiness. Each snippet focuses on the core idea.
Async Foundation
We use httpx.AsyncClient to issue non‑blocking GETs. The event loop schedules many tasks, freeing the CPU while waiting for network responses.
import asyncio
import httpx
async def fetch_html(url, client):
async with client.get(url, timeout=10.0) as resp:
resp.raise_for_status()
return await resp.text()
Rate Limiting & Concurrency Control
A semaphore caps the number of in‑flight requests. aiolimiter enforces a per‑second request cap and respects Retry‑After headers.
from aiolimiter import Limiter
async def fetch_with_limits(url, client, sem, limiter):
async with sem, limiter:
return await fetch_html(url, client)
Retries with Exponential Backoff
Tenacity decorates the fetch call. We retry on network errors, 5xx status codes, and 429s, adding jitter to avoid thundering herd.
from tenacity import retry, wait_exponential, stop_after_attempt, retry_if_exception_type, retry_if_result
@retry(
wait=wait_exponential(multiplier=1, min=1, max=60),
stop=stop_after_attempt(5),
retry=(
retry_if_exception_type(httpx.RequestError) |
retry_if_exception_type(httpx.HTTPStatusError) |
retry_if_result(lambda r: r.status_code in {429, 500, 502, 503, 504})
),
reraise=True
)
async def safe_fetch(url, client, sem, limiter):
return await fetch_with_limits(url, client, sem, limiter)
Simple Circuit Breaker
The circuit breaker tracks consecutive failures per host. After 3 failures it opens the circuit for 60 s, skipping requests to that host.
import time
class CircuitBreaker:
def __init__(self, max_failures=3, timeout=60):
self.max_failures = max_failures
self.timeout = timeout
self.states = {} # host -> (fail_count, open_until)
def should_skip(self, host):
state = self.states.get(host)
if state and state[1] and time.time() < state[1]:
return True
return False
def record_success(self, host):
self.states.pop(host, None)
def record_failure(self, host):
fail_count, _ = self.states.get(host, (0, None))
fail_count += 1
if fail_count >= self.max_failures:
self.states[host] = (fail_count, time.time() + self.timeout)
else:
self.states[host] = (fail_count, None)
Production Readiness: CLI, Logging, Metrics
We wire everything together with click, structured logging, and a progress bar. Metrics are collected in a shared dictionary and printed at the end.
import click
import logging
from tqdm.asyncio import tqdm
@click.command()
@click.option("--rss-url", default="https://discord.com/blog/rss.xml", help="RSS feed URL")
@click.option("--max-concurrency", default=100, type=int, help="Max concurrent HTTP requests")
@click.option("--rate-limit-rps", default=50.0, type=float, help="Global requests per second")
@click.option("--output-file", default="discord_blog_posts.jsonl", help="Output JSONL file")
@click.option("--fetch-limit", default=20, type=int, help="Number of article links to fetch")
@click.option("--log-level", default="INFO", type=click.Choice(["DEBUG", "INFO", "WARNING", "ERROR"]))
def main(rss_url, max_concurrency, rate_limit_rps, output_file, fetch_limit, log_level):
logging.basicConfig(level=log_level, format="%(asctime)s %(levelname)s %(message)s")
urls = load_rss(rss_url, fetch_limit)
stats = {"total": 0, "success": 0, "failed": 0, "retries": 0, "latencies": []}
circuit = CircuitBreaker()
asyncio.run(run_fetches(urls, max_concurrency, rate_limit_rps, output_file, stats, circuit))
print_stats(stats)
Putting It Together
The run_fetches coroutine orchestrates the async workflow: it creates a shared httpx.AsyncClient, a semaphore, an aiolimiter, and schedules tasks with tqdm.asyncio.tqdm. Each task records latency, updates stats, and writes a JSON line. Errors are caught and logged; failures after exhausting retries are marked with a special status.
async def run_fetches(urls, max_concurrency, rate_limit_rps, output_file, stats, circuit):
sem = asyncio.Semaphore(max_concurrency)
limiter = Limiter(rate_limit_rps, time_period=1)
async with httpx.AsyncClient(follow_redirects=True) as client:
with open(output_file, "w", encoding="utf-8") as fp:
tasks = []
for url in urls:
host = httpx.URL(url).host
if circuit.should_skip(host):
logging.warning(f"Circuit open for {host}, skipping {url}")
continue
task = asyncio.create_task(
process_url(url, client, sem, limiter, circuit, fp, stats)
)
tasks.append(task)
await tqdm.asyncio.tqdm.gather(*tasks, total=len(tasks))
Complete Script
The full runnable script combining all steps:
#!/usr/bin/env python3
import asyncio
import json
import logging
import time
from pathlib import Path
import feedparser
import httpx
import click
from bs4 import BeautifulSoup
from aiolimiter import Limiter
from tqdm.asyncio import tqdm
from tenacity import retry, wait_exponential, stop_after_attempt, retry_if_exception_type, retry_if_result
# ---------- Core helpers ----------
def load_rss(rss_url: str, fetch_limit: int):
feed = feedparser.parse(rss_url)
return [entry.link for entry in feed.entries[:fetch_limit]]
def extract_snippet(html: str, length: int = 400):
soup = BeautifulSoup(html, "html.parser")
text = soup.get_text(separator=" ", strip=True)
return text[:length]
class CircuitBreaker:
def __init__(self, max_failures: int = 3, timeout: int = 60):
self.max_failures = max_failures
self.timeout = timeout
self.states = {} # host -> (fail_count, open_until)
def should_skip(self, host: str) -> bool:
state = self.states.get(host)
if state and state[1] and time.time() < state[1]:
return True
return False
def record_success(self, host: str):
self.states.pop(host, None)
def record_failure(self, host: str):
fail_count, _ = self.states.get(host, (0, None))
fail_count += 1
if fail_count >= self.max_failures:
self.states[host] = (fail_count, time.time() + self.timeout)
else:
self.states[host] = (fail_count, None)
# ---------- Async fetch with retry ----------
@retry(
wait=wait_exponential(multiplier=1, min=1, max=60),
stop=stop_after_attempt(5),
retry=(
retry_if_exception_type(httpx.RequestError) |
retry_if_exception_type(httpx.HTTPStatusError) |
retry_if_result(lambda r: r.status_code in {429, 500, 502, 503, 504})
),
reraise=True
)
async def safe_fetch(url: str, client: httpx.AsyncClient, sem: asyncio.Semaphore, limiter: Limiter):
async with sem, limiter:
resp = await client.get(url, timeout=10.0)
resp.raise_for_status()
return resp
# ---------- Process single URL ----------
async def process_url(
url: str,
client: httpx.AsyncClient,
sem: asyncio.Semaphore,
limiter: Limiter,
circuit: CircuitBreaker,
fp,
stats: dict,
):
host = httpx.URL(url).host
start = time.perf_counter()
try:
resp = await safe_fetch(url, client, sem, limiter)
html = await resp.text()
snippet = extract_snippet(html)
record = {
"title": resp.headers.get("X-Title", url),
"url": url,
"extracted_text_snippet": snippet,
"publication_date": resp.headers.get("Date", ""),
"fetch_status": "SUCCESS",
"error_message": None,
"timestamp": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
}
fp.write(json.dumps(record) + "\n")
circuit.record_success(host)
stats["success"] += 1
except httpx.HTTPStatusError as e:
if e.response.status_code == 429:
retry_after = e.response.headers.get("Retry-After")
if retry_after:
await asyncio.sleep(int(retry_after))
logging.warning(f"HTTP error {e.response.status_code} on {url}")
record = {
"title": None,
"url": url,
"extracted_text_snippet": None,
"publication_date": None,
"fetch_status": "FAILED_RETRY_EXHAUSTED",
"error_message": str(e),
"timestamp": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
}
fp.write(json.dumps(record) + "\n")
circuit.record_failure(host)
stats["failed"] += 1
except httpx.RequestError as e:
logging.error(f"Request error on {url}: {e}")
record = {
"title": None,
"url": url,
"extracted_text_snippet": None,
"publication_date": None,
"fetch_status": "FAILED_RETRY_EXHAUSTED",
"error_message": str(e),
"timestamp": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
}
fp.write(json.dumps(record) + "\n")
circuit.record_failure(host)
stats["failed"] += 1
except Exception as e:
logging.exception(f"Unexpected error on {url}")
record = {
"title": None,
"url": url,
"extracted_text_snippet": None,
"publication_date": None,
"fetch_status": "FAILED_RETRY_EXHAUSTED",
"error_message": str(e),
"timestamp": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
}
fp.write(json.dumps(record) + "\n")
circuit.record_failure(host)
stats["failed"] += 1
finally:
latency = (time.perf_counter() - start) * 1000
stats["latencies"].append(latency)
stats["total"] += 1
# ---------- CLI and orchestration ----------
@click.command()
@click.option("--rss-url", default="https://discord.com/blog/rss.xml", help="RSS feed URL")
@click.option("--max-concurrency", default=100, type=int, help="Max concurrent HTTP requests")
@click.option("--rate-limit-rps", default=50.0, type=float, help="Global requests per second")
@click.option("--output-file", default="discord_blog_posts.jsonl", help="Output JSONL file")
@click.option("--fetch-limit", default=20, type=int, help="Number of article links to fetch")
@click.option("--log-level", default="INFO", type=click.Choice(["DEBUG", "INFO", "WARNING", "ERROR"]))
def main(rss_url, max_concurrency, rate_limit_rps, output_file, fetch_limit, log_level):
logging.basicConfig(level=log_level, format="%(asctime)s %(levelname)s %(message)s")
urls = load_rss(rss_url, fetch_limit)
stats = {"total": 0, "success": 0, "failed": 0, "latencies": []}
circuit = CircuitBreaker()
asyncio.run(run_fetches(urls, max_concurrency, rate_limit_rps, output_file, stats, circuit))
print_stats(stats)
async def run_fetches(urls, max_concurrency, rate_limit_rps, output_file, stats, circuit):
sem = asyncio.Semaphore(max_concurrency)
limiter = Limiter(rate_limit_rps, time_period=1)
async with httpx.AsyncClient(follow_redirects=True) as client:
with open(output_file, "w", encoding="utf-8") as fp:
tasks = [
asyncio.create_task(process_url(url, client, sem, limiter, circuit, fp, stats))
for url in urls
]
await tqdm.asyncio.tqdm.gather(*tasks, total=len(tasks))
def print_stats(stats):
avg_latency = sum(stats["latencies"]) / len(stats["latencies"]) if stats["latencies"] else 0
print("\n=== Summary ===")
print(f"Total requests: {stats['total']}")
print(f" Success: {stats['success']}")
print(f" Failed: {stats['failed']}")
print(f"Average latency: {avg_latency:.2f} ms")
if __name__ == "__main__":
main()
Expected Output
When you run python script.py --fetch-limit 10 you’ll see a progress bar, real‑time logs, and a final summary similar to:
2026-10-22 12:34:56 INFO fetching rss feed
2026-10-22 12:34:56 INFO starting 10 concurrent workers
2026-10-22 12:34:57 INFO fetched article: https://discord.com/blog/...
...
=== Summary ===
Total requests: 10
Success: 10
Failed: 0
Average latency: 123.45 ms
The file discord_blog_posts.jsonl will contain 10 JSON lines, each with the fields described in the specification.
Limitations and Tradeoffs
• Memory Footprint: Storing all latency samples and keeping the entire output file open can be heavy if fetch_limit scales to thousands.
• API Specificity: The retry rules and circuit breaker thresholds are tuned for Discord’s RSS feed; other APIs may require different status code handling or backoff strategies.
• Learning Curve: Async code with semaphores, limiters, and decorators can be harder to debug than a simple loop.
• Single‑Instance Scaling: This design runs on one machine. Coordinating rate limits across multiple instances would need an external lock or shared Redis counter.
• Rate‑Limit Granularity: aiolimiter applies a global RPS cap; per‑endpoint limits are not enforced unless you instantiate separate limiters.
Frequently Asked Questions
How do I determine the optimal max_concurrency and rate_limit_rps for an API without official documentation?
Start with a small concurrency (e.g., 10) and gradually increase while monitoring latency and error rates. Use curl -w %{time_total} or the script’s latency metrics to spot saturation. If you hit many 429s, reduce rate_limit_rps or implement per‑host limiters.
When should I choose asyncio over ThreadPoolExecutor or ProcessPoolExecutor for I/O‑bound tasks?
Asyncio shines when the work is primarily network I/O and you need to keep the CPU free for many concurrent connections. Thread pools are simpler but still block the thread on each I/O call, limiting concurrency to the number of threads. Process pools are overkill for I/O‑bound workloads.
How can I securely handle API authentication (e.g., OAuth2, API keys) in a high‑throughput CLI?
Store credentials in environment variables or a secrets manager, inject them into httpx.AsyncClient via auth or headers, and rotate tokens if the API supports it. Avoid hard‑coding secrets in the script.
What are common pitfalls when debugging asyncio applications, especially with third‑party libraries?
1. Forgetting to await an async function.
2. Mixing sync and async code without an event loop.
3. Not handling asyncio.CancelledError in long‑running tasks.
4. Using blocking calls (e.g., time.sleep) inside async functions; use await asyncio.sleep instead.
How can I integrate this CLI into a larger data pipeline (e.g., scheduled jobs, data warehousing)?
Wrap the CLI as a Docker container or a scheduled cron job. Emit structured logs that a log aggregator can parse. Write the JSONL to a cloud storage bucket or stream it into a message queue for downstream consumers.
What I'd Change
In production, I’d replace the in‑memory circuit breaker with a Redis‑backed one so that multiple workers share state. I’d also move the rate limiter to a token‑bucket service that respects per‑host limits, and I’d instrument the CLI with Prometheus metrics so ops can monitor request rates, error rates, and latency in real time. Finally, I’d add a graceful shutdown hook to ensure the output file is flushed and partial results are not lost if the process is killed mid‑run.