Python Async Cheat Sheet
asyncio for people calling APIs in parallel
Most AI code is IO-bound: waiting on model APIs. asyncio is how one process waits on fifty calls at once instead of one at a time.
Running many at once
| Construct | Behaviour |
|---|---|
await asyncio.gather(*tasks)
|
All in parallel, results in input order |
await asyncio.gather(*tasks, return_exceptions=True)
|
One failure no longer cancels the rest |
async with asyncio.TaskGroup() as tg:
|
Structured concurrency — 3.11+, cancels siblings on error |
for c in asyncio.as_completed(tasks):
|
Results as they finish, not in order |
await asyncio.wait_for(coro, timeout=30)
|
Per-call deadline |
asyncio.Semaphore(10)
|
Cap in-flight calls — respects rate limits |
await asyncio.to_thread(blocking_fn)
|
Move a blocking call off the event loop |
The traps
| Mistake | Symptom | Fix |
|---|---|---|
Calling a coroutine without await
|
A RuntimeWarning and nothing runs | await it, or wrap in create_task |
A blocking call inside async code
|
Everything serialises and async gains nothing | to_thread, or an async client |
time.sleep() in a coroutine
|
The whole event loop stops | await asyncio.sleep() |
Unbounded gather over 10k items
|
Rate limits, memory, and a thundering herd | Semaphore or chunk the list |
Fire-and-forget create_task with no reference
|
The task is garbage-collected mid-flight | Keep the reference until it completes |
Code examples
Bounded parallel API calls
The pattern almost every AI pipeline needs: run many requests at once, but never more than N in flight.
import asyncio
async def fetch(client, prompt, sem):
async with sem: # never more than N in flight
return await client.complete(prompt)
async def main(client, prompts, limit=10):
sem = asyncio.Semaphore(limit)
tasks = [fetch(client, p, sem) for p in prompts]
# return_exceptions keeps one failure from cancelling the rest
results = await asyncio.gather(*tasks, return_exceptions=True)
ok = [r for r in results if not isinstance(r, Exception)]
bad = [r for r in results if isinstance(r, Exception)]
print(f"{len(ok)} ok, {len(bad)} failed")
return ok
asyncio.run(main(client, prompts))Frequently asked questions
Will asyncio make my code faster?
Only if it is IO-bound — waiting on APIs, disks or sockets. CPU-bound work still runs on one thread and gains nothing; that needs multiprocessing.
Why did adding async change nothing?
Almost certainly a blocking call inside a coroutine, which stalls the whole event loop. Move it with asyncio.to_thread, or swap in an async client.
Was this cheat sheet useful?
Comments
No comments yet — be the first.
Keep going
Browse all
More cheat sheets
Need a different cheat sheet?
Tell us what you would like to see and we will build it — free.
Request a cheat sheet
