Python AsyncIO Cheat Sheet
Covers async/await syntax, creating and gathering concurrent tasks, synchronization primitives, and timeout/error handling patterns.
2 PagesAdvancedMar 25, 2026
Coroutines & await
Define and run coroutine functions.
python
import asyncioasync def fetch_data(delay): await asyncio.sleep(delay) # Non-blocking sleep return f"data after {delay}s"async def main(): result = await fetch_data(1) print(result)asyncio.run(main()) # Entry point: creates and runs the event loop
Tasks & Concurrent Execution
Run multiple coroutines concurrently.
python
import asyncioasync def main(): # Schedule coroutines to run concurrently task1 = asyncio.create_task(fetch_data(1)) task2 = asyncio.create_task(fetch_data(2)) result1 = await task1 result2 = await task2 # Or gather many at once, preserving input order in the results results = await asyncio.gather( fetch_data(1), fetch_data(2), fetch_data(3) ) # Wait for just the first to finish done, pending = await asyncio.wait( [task1, task2], return_when=asyncio.FIRST_COMPLETED )asyncio.run(main())
Synchronization Primitives
Coordinate access between coroutines.
python
import asynciolock = asyncio.Lock()sem = asyncio.Semaphore(3) # Limit to 3 concurrent tasksasync def worker(n): async with sem: # Acquire/release semaphore async with lock: # Acquire/release mutex print(f"worker {n} running") await asyncio.sleep(1)queue = asyncio.Queue()async def producer(): for i in range(5): await queue.put(i)async def consumer(): while True: item = await queue.get() print(f"consumed {item}") queue.task_done()
Timeouts & Error Handling
Guard against slow or failing coroutines.
python
import asyncioasync def main(): try: result = await asyncio.wait_for(fetch_data(5), timeout=2) except asyncio.TimeoutError: print("Timed out!") # gather() with exceptions captured instead of raised results = await asyncio.gather( fetch_data(1), bad_coro(), return_exceptions=True ) for r in results: if isinstance(r, Exception): print("failed:", r)
Key Concepts
Core vocabulary for asyncio code.
- async def- Defines a coroutine function; calling it returns a coroutine object, not the result
- await- Suspends the coroutine until the awaited object completes, yielding control to the event loop
- asyncio.create_task()- Schedules a coroutine to run concurrently on the event loop, returns immediately
- asyncio.run()- Creates a new event loop, runs the coroutine to completion, and closes the loop
- Event loop- The single-threaded scheduler that runs coroutines/tasks and dispatches I/O callbacks
- async with / async for- Async context managers and iterators for use inside coroutines
Structured Concurrency with TaskGroup
Python 3.11+ replaces manual gather() bookkeeping with automatic cancellation on error.
python
import asyncioasync def main(): async with asyncio.TaskGroup() as tg: task1 = tg.create_task(fetch_data(1)) task2 = tg.create_task(fetch_data(2)) task3 = tg.create_task(fetch_data(3)) # All tasks are awaited here; if any raised, the others are # cancelled and an ExceptionGroup is raised from the `async with` print(task1.result(), task2.result(), task3.result())asyncio.run(main())# Handling the ExceptionGroup (3.11+)try: asyncio.run(main())except* ValueError as eg: for exc in eg.exceptions: print("failed:", exc)
Cancellation & shield()
Control which coroutines survive a cancellation request.
python
import asyncioasync def critical_write(): await asyncio.sleep(1) return "saved"async def main(): task = asyncio.create_task(fetch_data(10)) await asyncio.sleep(0.1) task.cancel() # Requests cancellation try: await task except asyncio.CancelledError: print("task was cancelled") outer = asyncio.current_task() try: # Protects critical_write() from being cancelled if `outer` is result = await asyncio.shield(critical_write()) except asyncio.CancelledError: print("outer cancelled, but write kept running")
Offloading Blocking / CPU-bound Work
Run synchronous or CPU-heavy code without blocking the event loop.
python
import asynciofrom concurrent.futures import ProcessPoolExecutordef cpu_heavy(n): return sum(i * i for i in range(n))def blocking_io(): with open("file.txt") as f: return f.read()async def main(): loop = asyncio.get_running_loop() # Default executor (thread pool) for blocking I/O text = await loop.run_in_executor(None, blocking_io) # Process pool for CPU-bound work (bypasses the GIL) with ProcessPoolExecutor() as pool: result = await loop.run_in_executor(pool, cpu_heavy, 10_000_000) # asyncio.to_thread() is the shorthand for thread offload (3.9+) text2 = await asyncio.to_thread(blocking_io)
Streams: TCP Client/Server
Low-level asyncio networking without a full framework.
python
import asyncioasync def handle_client(reader, writer): data = await reader.read(100) addr = writer.get_extra_info("peername") writer.write(data.upper()) await writer.drain() writer.close() await writer.wait_closed()async def run_server(): server = await asyncio.start_server(handle_client, "127.0.0.1", 8888) async with server: await server.serve_forever()async def run_client(): reader, writer = await asyncio.open_connection("127.0.0.1", 8888) writer.write(b"hello\n") await writer.drain() response = await reader.read(100) writer.close()
Advanced Coordination APIs
Less common but essential primitives for real systems.
- asyncio.Event()- One-to-many signal; waiters unblock when set() is called
- asyncio.Condition()- Lock + notify/notify_all for producer-consumer coordination
- asyncio.Barrier(n)- Blocks n tasks until all n have reached the barrier (3.11+)
- contextvars.ContextVar- Per-task context propagation (e.g. request IDs) that survives across await points
- loop.call_later() / call_at()- Schedule a callback after a delay or at an absolute loop time
- asyncio.as_completed()- Iterate awaitables in completion order rather than input order
- uvloop- Drop-in libuv-based event loop replacement, 2-4x faster than the default
Pro Tip
asyncio gives concurrency, not parallelism — it's built for I/O-bound work like network and disk calls. CPU-bound code still blocks the single-threaded event loop; offload it with loop.run_in_executor() or a ProcessPoolExecutor.
Was this cheat sheet helpful?
Explore Topics
#PythonAsyncIO#PythonAsyncIOCheatSheet#Programming#Advanced#CoroutinesAwait#TasksConcurrentExecution#SynchronizationPrimitives#TimeoutsErrorHandling#Concurrency#ErrorHandling#CheatSheet#SkillVeris
