Real-world asyncio recipes for HTTP concurrency, retries, and pipelines.
Keep high throughput without overwhelming downstream services.
import asyncio
sem = asyncio.Semaphore(5)
async def work(item):
async with sem:
await asyncio.sleep(0.2)
return item * 10
async def main():
results = await asyncio.gather(*(work(i) for i in range(20)))
print(results[:5])
asyncio.run(main())A standard pattern for APIs, crawlers, and batch processors.
import asyncio
async def process(x):
await asyncio.sleep(0.05)
return x * 2
async def run_batch(items, size=10):
results = []
for i in range(0, len(items), size):
batch = items[i:i+size]
results.extend(await asyncio.gather(*(process(x) for x in batch)))
return results
print(asyncio.run(run_batch(list(range(25)))))Batches help when you want memory control or stricter request pacing.
Add retries, backoff, and cancellation-aware wrappers.
import asyncio
attempts = 0
async def flaky():
global attempts
attempts += 1
if attempts < 3:
raise RuntimeError("temporary failure")
return "ok"
async def with_retry():
delay = 0.2
for _ in range(5):
try:
return await flaky()
except RuntimeError:
await asyncio.sleep(delay)
delay *= 2
raise RuntimeError("exhausted retries")
print(asyncio.run(with_retry()))Keep retries bounded and cancellation-aware in production systems.
import asyncio
async def worker():
try:
while True:
await asyncio.sleep(1)
except asyncio.CancelledError:
print("worker shutting down")
raise
async def main():
task = asyncio.create_task(worker())
await asyncio.sleep(0.2)
task.cancel()
await asyncio.gather(task, return_exceptions=True)
asyncio.run(main())Use this pattern to ensure graceful shutdown in services and daemons.
Inspect running tasks and enable asyncio debug support.
import asyncio
async def sleeper():
await asyncio.sleep(1)
async def main():
task = asyncio.create_task(sleeper(), name="demo-task")
for t in asyncio.all_tasks():
print(t.get_name())
await task
asyncio.run(main())Use task introspection when debugging leaked or stuck coroutines.
import asyncio
async def main():
loop = asyncio.get_running_loop()
loop.set_debug(True)
await asyncio.sleep(0.1)
asyncio.run(main(), debug=True)Debug mode helps surface slow callbacks, resource warnings, and some misuse patterns.