Ever watched a simple “fetch‑all‑posts” script grind to a halt because the API hiccups for a second? You’re not alone—our team’s nightly sync of JSONPlaceholder data stalled whenever the public service slowed, and the whole pipeline missed its SLA. If you need a fast, fault‑tolerant fetcher that can hammer many endpoints in parallel, respect a concurrency budget, and exit gracefully when the host restarts, this tutorial is for you. I’ll replace the blocking requests loop with an async, retry‑aware client and show you exactly how to ship it to production.
Key Takeaways
- Asyncio + aiohttp + a semaphore gives you controllable parallelism without overwhelming the remote service.
- Exponential backoff with jitter turns transient HTTP failures into harmless retries.
- Signal‑driven graceful shutdown prevents data loss and leaves the event loop in a clean state.
- Persisting each record as JSON‑Lines while streaming keeps memory usage constant.
- Structured logging + simple counters give you visibility without pulling in a full observability stack.
The Problem
Our original pipeline used a synchronous requests.get inside a for loop over post IDs 1‑100. When the JSONPlaceholder service responded slowly (or returned 5xx), the loop blocked for seconds, causing the nightly job to exceed its 5‑minute window. Moreover, the script had no way to limit concurrent connections, so a sudden spike in retries could inadvertently trigger rate‑limit bans. Finally, a Ctrl‑C during execution left the output file half‑written, forcing manual cleanup.
Data and Sources
We pull real posts from the public JSONPlaceholder API (https://jsonplaceholder.typicode.com/). The official docs, aiohttp docs, and asyncio docs are the primary references.
Data accessed on 2026-11-01.
Building the Async Client
Step 1 — Setup Async Environment and aiohttp Session
The first sub‑problem is creating a reusable, non‑blocking HTTP client that can be shared across many coroutines.
import aiohttp
import asyncio
async def create_session():
timeout = aiohttp.ClientTimeout(total=10)
return aiohttp.ClientSession(timeout=timeout)
We configure a total timeout of 10 seconds to avoid hanging forever. The session is created once and passed to every worker.
Step 2 — Concurrency Control with Semaphore
We need to bound the number of simultaneous requests (e.g., 10) to stay under the API’s informal limit.
MAX_CONCURRENT = 10
semaphore = asyncio.Semaphore(MAX_CONCURRENT)
async def fetch_with_semaphore(session, url):
async with semaphore:
return await fetch_one(session, url)
The semaphore guarantees that at most MAX_CONCURRENT coroutines enter the network block at any time.
Step 3 — Exponential Backoff with Jitter
Transient failures (HTTP 5xx, timeouts) should be retried, but we must back off to give the server breathing room.
import random
import async_timeout
MAX_RETRIES = 5
BASE_DELAY = 0.5 # seconds
async def fetch_one(session, url):
for attempt in range(1, MAX_RETRIES + 1):
try:
async with async_timeout.timeout(5):
async with session.get(url) as resp:
resp.raise_for_status()
return await resp.json()
except (aiohttp.ClientError, asyncio.TimeoutError) as exc:
if attempt == MAX_RETRIES:
raise
delay = BASE_DELAY * (2 ** (attempt - 1))
jitter = random.uniform(0, delay * 0.1)
await asyncio.sleep(delay + jitter)
The loop retries up to MAX_RETRIES times, doubling the delay each attempt and adding a small random jitter to avoid thundering‑herd effects.
Step 4 — Graceful Shutdown on SIGINT / SIGTERM
If the process receives a termination signal, we must cancel pending tasks, close the session, and flush any buffered writes.
import signal
def shutdown(loop, session):
for task in asyncio.all_tasks(loop):
task.cancel()
loop.run_until_complete(session.close())
def install_signal_handlers(loop, session):
for sig in (signal.SIGINT, signal.SIGTERM):
loop.add_signal_handler(sig, lambda: shutdown(loop, session))
Signal handlers cancel all running tasks and close the aiohttp session, ensuring no dangling connections.
Step 5 — Persist Results to a JSON Lines File
Streaming each successful fetch to a .jsonl file keeps memory usage constant and makes downstream processing trivial.
import aiofiles
OUTPUT_PATH = "posts.jsonl"
async def write_record(file, record):
line = json.dumps(record) + "\n"
await file.write(line)
async def worker(session, post_id, file):
url = f"https://jsonplaceholder.typicode.com/posts/{post_id}"
try:
data = await fetch_with_semaphore(session, url)
await write_record(file, data)
logger.info("Fetched %s", post_id)
except Exception as exc:
logger.error("Failed %s: %s", post_id, exc)
Each worker writes its result immediately, so a crash never loses already‑fetched data.
Step 6 — Structured Logging & Basic Metrics
Simple counters let us see how many posts succeeded or failed without pulling in a full metrics system.
import logging
import json
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s %(levelname)s %(message)s",
)
logger = logging.getLogger("async_fetcher")
success_counter = 0
failure_counter = 0
Inside worker we increment these counters; after the run we log a summary.
Putting It All Together
The orchestrator builds the session, opens the output file, spawns a coroutine per post ID, and waits for completion while respecting shutdown signals.
async def main(post_ids):
session = await create_session()
loop = asyncio.get_running_loop()
install_signal_handlers(loop, session)
async with aiofiles.open(OUTPUT_PATH, mode="w") as file:
tasks = [
asyncio.create_task(worker(session, pid, file))
for pid in post_ids
]
await asyncio.gather(*tasks, return_exceptions=True)
await session.close()
logger.info("Run complete – success: %d, failures: %d", success_counter, failure_counter)
if __name__ == "__main__":
ids = list(range(1, 101)) # 1‑100 inclusive
asyncio.run(main(ids))
We use asyncio.gather(..., return_exceptions=True) so a single failure doesn’t cancel the whole batch.
Complete Script
The full runnable script combining all steps:
#!/usr/bin/env python3
"""
Async/await in Production: Resilient concurrent fetcher for JSONPlaceholder.
"""
import asyncio
import aiohttp
import aiofiles
import async_timeout
import json
import logging
import random
import signal
# ------------------- Configuration -------------------
MAX_CONCURRENT = 10
MAX_RETRIES = 5
BASE_DELAY = 0.5 # seconds
OUTPUT_PATH = "posts.jsonl"
POST_IDS = list(range(1, 101)) # 1‑100 inclusive
# -----------------------------------------------------
# Structured logging
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s %(levelname)s %(message)s",
)
logger = logging.getLogger("async_fetcher")
# Simple metrics
success_counter = 0
failure_counter = 0
async def create_session() -> aiohttp.ClientSession:
"""Create a single aiohttp session with a sensible timeout."""
timeout = aiohttp.ClientTimeout(total=10)
return aiohttp.ClientSession(timeout=timeout)
async def fetch_one(session: aiohttp.ClientSession, url: str) -> dict:
"""Fetch a URL with exponential backoff + jitter."""
global success_counter, failure_counter
for attempt in range(1, MAX_RETRIES + 1):
try:
async with async_timeout.timeout(5):
async with session.get(url) as resp:
resp.raise_for_status()
data = await resp.json()
success_counter += 1
return data
except (aiohttp.ClientError, asyncio.TimeoutError) as exc:
if attempt == MAX_RETRIES:
failure_counter += 1
raise
delay = BASE_DELAY * (2 ** (attempt - 1))
jitter = random.uniform(0, delay * 0.1)
await asyncio.sleep(delay + jitter)
semaphore = asyncio.Semaphore(MAX_CONCURRENT)
async def fetch_with_semaphore(session: aiohttp.ClientSession, url: str) -> dict:
"""Wrap fetch_one with a concurrency limiter."""
async with semaphore:
return await fetch_one(session, url)
async def write_record(file: aiofiles.threadpool.binary.AsyncBufferedWriter, record: dict):
"""Append a single JSON object as a line."""
line = json.dumps(record) + "\n"
await file.write(line)
async def worker(session: aiohttp.ClientSession, post_id: int, file: aiofiles.threadpool.binary.AsyncBufferedWriter):
"""Fetch one post and persist it; log successes and failures."""
url = f"https://jsonplaceholder.typicode.com/posts/{post_id}"
try:
data = await fetch_with_semaphore(session, url)
await write_record(file, data)
logger.info("Fetched post %s", post_id)
except Exception as exc:
logger.error("Failed post %s: %s", post_id, exc)
def shutdown(loop: asyncio.AbstractEventLoop, session: aiohttp.ClientSession):
"""Cancel all tasks and close the aiohttp session."""
logger.info("Shutdown signal received – cancelling tasks")
for task in asyncio.all_tasks(loop):
task.cancel()
loop.run_until_complete(session.close())
def install_signal_handlers(loop: asyncio.AbstractEventLoop, session: aiohttp.ClientSession):
"""Register SIGINT and SIGTERM handlers for graceful termination."""
for sig in (signal.SIGINT, signal.SIGTERM):
loop.add_signal_handler(sig, lambda: shutdown(loop, session))
async def main(post_ids):
"""Orchestrate the whole pipeline."""
session = await create_session()
loop = asyncio.get_running_loop()
install_signal_handlers(loop, session)
async with aiofiles.open(OUTPUT_PATH, mode="w") as file:
tasks = [
asyncio.create_task(worker(session, pid, file))
for pid in post_ids
]
# Gather with return_exceptions to let other tasks finish even if one fails
await asyncio.gather(*tasks, return_exceptions=True)
await session.close()
logger.info(
"Run complete – success: %d, failures: %d",
success_counter,
failure_counter,
)
if __name__ == "__main__":
asyncio.run(main(POST_IDS))
Expected Output
Running python async_fetcher.py creates posts.jsonl with 100 lines, each a valid JSON object. The console shows timestamps, e.g.:
2026-11-01 12:00:01,234 INFO Fetched post 1
2026-11-01 12:00:01,236 INFO Fetched post 2
...
2026-11-01 12:00:05,789 INFO Run complete – success: 100, failures: 0
If a few requests time out, you’ll see ERROR lines but the script still finishes, and the output file contains every successful record.
Limitations and Tradeoffs
Asyncio adds mental overhead: stack traces are less familiar, and debugging requires tools like asyncio.run introspection or aiomonitor. The semaphore caps concurrency but does not enforce a true rate limit; if the API changes its quota policy, you may need a token‑bucket limiter (e.g., aiolimiter). The script assumes JSON responses; binary payloads would need await resp.read() and a different write strategy. Finally, the simple counters live in the process memory; scaling this to multiple workers would require an external metrics backend.
Frequently Asked Questions
Can I reuse this pattern for APIs that require authentication?
Yes. Add the required headers (e.g., Authorization: Bearer …) when constructing the ClientSession or pass them per request.
What if the API imposes a per‑minute request quota instead of a concurrent limit?
Replace the semaphore with an async token‑bucket limiter such as aiolimiter, which releases a fixed number of tokens each minute.
How do I collect richer metrics like latency histograms?
Wrap the fetch_one call with timing logic (e.g., time.perf_counter()) and push the values to a Prometheus Histogram or OpenTelemetry exporter.
Will this approach work on Windows?
Signal handling differs on Windows; loop.add_signal_handler raises NotImplementedError. On Windows you can skip installing signal handlers or use the win32api package for graceful termination.
Is writing each line with aiofiles safe for high‑throughput workloads?
For very high write rates you might batch lines in memory and flush periodically, or switch to a fast binary format (e.g., Apache Arrow) and a dedicated writer.
What I'd Change
If I were to ship this to a larger team, I’d replace the ad‑hoc counters with a Prometheus client, move the retry/backoff logic into a reusable decorator, and add a configurable token‑bucket limiter to honor explicit rate‑limit headers. Those tweaks turn a solid prototype into a production‑ready component that can survive API changes, scale horizontally, and integrate with our observability stack.