|$ curl https://forge-ai.dev/api/markdown?path=docs/python/asyncio
$cat docs/python-—-asyncio.md
updated Recently·22 min read·published
Python — Asyncio
◆Python◆Advanced
Introduction
Asyncio is Python's library for asynchronous I/O using coroutines. It uses cooperative multitasking — tasks voluntarily yield control at await points, enabling high concurrency without threads or processes.
Unlike threading (preemptive, GIL-limited) or multiprocessing (parallel but heavy), asyncio runs on a single thread with an event loop that switches between tasks when they wait for I/O.
Coroutines & await
coroutines.py
Python
| 1 | # Define a coroutine with async def |
| 2 | async def fetch_data(url: str) -> str: |
| 3 | # await suspends this coroutine until result ready |
| 4 | response = await http_get(url) |
| 5 | return response.text |
| 6 | |
| 7 | # Calling a coroutine function returns a coroutine object |
| 8 | coro = fetch_data("https://example.com") |
| 9 | print(coro) # <coroutine object fetch_data at 0x...> |
| 10 | |
| 11 | # Running a coroutine requires an event loop |
| 12 | result = await coro # only works inside another coroutine |
| 13 | |
| 14 | # Run from sync code: |
| 15 | import asyncio |
| 16 | result = asyncio.run(fetch_data("https://example.com")) |
| 17 | # asyncio.run() creates event loop, runs coroutine, closes loop |
| 18 | |
| 19 | # Multiple awaitable types: |
| 20 | # - Coroutines (async def functions) |
| 21 | # - Tasks (wrapped coroutines running in background) |
| 22 | # - Futures (low-level awaitable) |
| 23 | # - asyncio.sleep(), asyncio.gather(), etc. |
Event Loop
event_loop.py
Python
| 1 | # asyncio.run() — main entry point (Python 3.7+) |
| 2 | async def main(): |
| 3 | await asyncio.sleep(1) |
| 4 | return "done" |
| 5 | |
| 6 | result = asyncio.run(main()) |
| 7 | |
| 8 | # Manual loop control (for frameworks) |
| 9 | loop = asyncio.new_event_loop() |
| 10 | asyncio.set_event_loop(loop) |
| 11 | try: |
| 12 | result = loop.run_until_complete(main()) |
| 13 | finally: |
| 14 | loop.close() |
| 15 | |
| 16 | # Getting the current loop |
| 17 | loop = asyncio.get_running_loop() # preferred (3.7+) |
| 18 | # asyncio.get_event_loop() # older API, avoid |
| 19 | |
| 20 | # Loop methods |
| 21 | loop = asyncio.get_running_loop() |
| 22 | loop.time() # current loop time (float) |
| 23 | loop.call_soon(lambda: print("immediate")) |
| 24 | loop.call_later(1, lambda: print("after 1s")) |
| 25 | loop.call_at(loop.time() + 2, lambda: print("at time+2")) |
Tasks & gather
tasks.py
Python
| 1 | # Tasks run coroutines concurrently |
| 2 | async def fetch(url: str) -> str: |
| 3 | await asyncio.sleep(1) # simulate I/O |
| 4 | return f"data from {url}" |
| 5 | |
| 6 | async def main(): |
| 7 | # create_task schedules the coroutine on the event loop |
| 8 | task1 = asyncio.create_task(fetch("url1")) |
| 9 | task2 = asyncio.create_task(fetch("url2")) |
| 10 | task3 = asyncio.create_task(fetch("url3")) |
| 11 | |
| 12 | # Tasks run in background, awaiting gets their result |
| 13 | result1 = await task1 |
| 14 | result2 = await task2 |
| 15 | result3 = await task3 |
| 16 | print(result1, result2, result3) # all take ~1s total |
| 17 | |
| 18 | # asyncio.gather — run multiple coroutines concurrently |
| 19 | async def main_gather(): |
| 20 | results = await asyncio.gather( |
| 21 | fetch("url1"), |
| 22 | fetch("url2"), |
| 23 | fetch("url3"), |
| 24 | ) |
| 25 | print(results) # ["data from url1", "data from url2", ...] |
| 26 | |
| 27 | # gather with return_exceptions |
| 28 | async def might_fail(n): |
| 29 | if n == 2: |
| 30 | raise ValueError("nope") |
| 31 | return n |
| 32 | |
| 33 | async def main_safe(): |
| 34 | results = await asyncio.gather( |
| 35 | might_fail(1), might_fail(2), might_fail(3), |
| 36 | return_exceptions=True, |
| 37 | ) |
| 38 | print(results) # [1, ValueError('nope'), 3] |
| 39 | |
| 40 | # as_completed — process results as they arrive |
| 41 | async def main_as_completed(): |
| 42 | tasks = [fetch(f"url_{i}") for i in range(5)] |
| 43 | for coro in asyncio.as_completed(tasks): |
| 44 | result = await coro |
| 45 | print(f"got: {result}") |
Async Context Managers
async_context.py
Python
| 1 | # Class-based async context manager |
| 2 | class AsyncDatabase: |
| 3 | async def __aenter__(self): |
| 4 | print("connecting to db...") |
| 5 | await asyncio.sleep(0.1) |
| 6 | self.conn = "db connection" |
| 7 | return self |
| 8 | |
| 9 | async def __aexit__(self, exc_type, exc_val, exc_tb): |
| 10 | print("closing connection...") |
| 11 | await asyncio.sleep(0.1) |
| 12 | self.conn = None |
| 13 | |
| 14 | async def query(self, sql: str) -> str: |
| 15 | await asyncio.sleep(0.05) |
| 16 | return f"result of {sql}" |
| 17 | |
| 18 | async def main(): |
| 19 | async with AsyncDatabase() as db: |
| 20 | result = await db.query("SELECT 1") |
| 21 | print(result) |
| 22 | |
| 23 | # @asynccontextmanager decorator |
| 24 | from contextlib import asynccontextmanager |
| 25 | |
| 26 | @asynccontextmanager |
| 27 | async def db_session(): |
| 28 | print("acquire") |
| 29 | conn = await asyncio.sleep(0.1, "conn") |
| 30 | try: |
| 31 | yield conn |
| 32 | finally: |
| 33 | print("release") |
| 34 | |
| 35 | async def main(): |
| 36 | async with db_session() as conn: |
| 37 | print(f"using {conn}") |
Async Iterators & Generators
async_iter.py
Python
| 1 | # Async iterator |
| 2 | class AsyncCounter: |
| 3 | def __init__(self, limit: int): |
| 4 | self.limit = limit |
| 5 | self.n = 0 |
| 6 | |
| 7 | def __aiter__(self): |
| 8 | return self |
| 9 | |
| 10 | async def __anext__(self): |
| 11 | if self.n >= self.limit: |
| 12 | raise StopAsyncIteration |
| 13 | await asyncio.sleep(0.1) |
| 14 | self.n += 1 |
| 15 | return self.n |
| 16 | |
| 17 | async def main(): |
| 18 | async for num in AsyncCounter(5): |
| 19 | print(num) # 1, 2, 3, 4, 5 |
| 20 | |
| 21 | # Async generator (Python 3.6+) |
| 22 | async def async_range(n: int): |
| 23 | for i in range(n): |
| 24 | await asyncio.sleep(0.1) |
| 25 | yield i |
| 26 | |
| 27 | async def main(): |
| 28 | async for num in async_range(5): |
| 29 | print(num) |
| 30 | |
| 31 | # Async list comprehension |
| 32 | async def main(): |
| 33 | results = [x async for x in async_range(5)] |
| 34 | print(results) # [0, 1, 2, 3, 4] |
Async Queue
async_queue.py
Python
| 1 | # Producer/Consumer pattern with asyncio.Queue |
| 2 | import asyncio |
| 3 | import random |
| 4 | |
| 5 | async def producer(queue: asyncio.Queue, n: int): |
| 6 | for i in range(n): |
| 7 | item = f"item_{i}" |
| 8 | await queue.put(item) |
| 9 | print(f"produced {item}") |
| 10 | await asyncio.sleep(random.random() * 0.5) |
| 11 | await queue.put(None) # sentinel — signals end |
| 12 | |
| 13 | async def consumer(queue: asyncio.Queue, name: str): |
| 14 | while True: |
| 15 | item = await queue.get() |
| 16 | if item is None: |
| 17 | queue.task_done() |
| 18 | break |
| 19 | print(f" {name} consumed {item}") |
| 20 | await asyncio.sleep(random.random() * 0.3) |
| 21 | queue.task_done() |
| 22 | |
| 23 | async def main(): |
| 24 | queue = asyncio.Queue(maxsize=5) |
| 25 | # Start producer and consumers concurrently |
| 26 | await asyncio.gather( |
| 27 | producer(queue, 10), |
| 28 | consumer(queue, "A"), |
| 29 | consumer(queue, "B"), |
| 30 | ) |
| 31 | await queue.join() |
| 32 | |
| 33 | asyncio.run(main()) |
ℹ
info
The sentinel pattern (None in the queue) is a clean way to signal consumers to stop. For multiple consumers, you need one sentinel per consumer, or use asyncio.Queue.shutdown() (Python 3.13+).
Running Blocking Code
sync_bridge.py
Python
| 1 | # run_in_executor — run blocking code in thread pool |
| 2 | import time |
| 3 | import requests |
| 4 | |
| 5 | def blocking_api_call(url: str) -> dict: |
| 6 | time.sleep(2) # blocking I/O |
| 7 | return {"url": url, "status": 200} |
| 8 | |
| 9 | async def main(): |
| 10 | loop = asyncio.get_running_loop() |
| 11 | result = await loop.run_in_executor( |
| 12 | None, # None = default ThreadPoolExecutor |
| 13 | blocking_api_call, |
| 14 | "https://api.example.com", |
| 15 | ) |
| 16 | print(result) |
| 17 | |
| 18 | # With concurrent.futures.ProcessPoolExecutor for CPU |
| 19 | from concurrent.futures import ProcessPoolExecutor |
| 20 | |
| 21 | def cpu_heavy(n: int) -> int: |
| 22 | return sum(i ** 2 for i in range(n)) |
| 23 | |
| 24 | async def main_cpu(): |
| 25 | loop = asyncio.get_running_loop() |
| 26 | with ProcessPoolExecutor() as pool: |
| 27 | result = await loop.run_in_executor( |
| 28 | pool, cpu_heavy, 10_000_000 |
| 29 | ) |
| 30 | print(result) |
| 31 | |
| 32 | # asyncio.to_thread (Python 3.9+) — simpler API |
| 33 | async def main_simple(): |
| 34 | result = await asyncio.to_thread(blocking_api_call, "https://example.com") |
| 35 | print(result) |
✓
best practice
Never call blocking code directly inside a coroutine — it blocks the entire event loop. Use run_in_executor or asyncio.to_thread for blocking I/O, database queries, or CPU-heavy work.
Synchronization Primitives
synchronization.py
Python
| 1 | # Asyncio provides thread-safe sync primitives |
| 2 | |
| 3 | # Lock — mutual exclusion |
| 4 | lock = asyncio.Lock() |
| 5 | async def critical_section(): |
| 6 | async with lock: |
| 7 | # only one coroutine at a time |
| 8 | await asyncio.sleep(0.1) |
| 9 | |
| 10 | # Semaphore — limit concurrent access |
| 11 | sem = asyncio.Semaphore(5) |
| 12 | async def limited(): |
| 13 | async with sem: |
| 14 | await asyncio.sleep(1) |
| 15 | |
| 16 | # Event — signal between coroutines |
| 17 | event = asyncio.Event() |
| 18 | async def waiter(): |
| 19 | print("waiting...") |
| 20 | await event.wait() |
| 21 | print("got signal!") |
| 22 | async def signaler(): |
| 23 | await asyncio.sleep(1) |
| 24 | event.set() # wake up all waiters |
| 25 | |
| 26 | # Condition — like Event but with notify |
| 27 | condition = asyncio.Condition() |
| 28 | async def consumer(): |
| 29 | async with condition: |
| 30 | await condition.wait() |
| 31 | print("notified") |
| 32 | async def producer(): |
| 33 | await asyncio.sleep(1) |
| 34 | async with condition: |
| 35 | condition.notify_all() |
Timeouts & Cancellation
timeouts.py
Python
| 1 | # asyncio.wait_for — timeout with exception |
| 2 | async def slow(): |
| 3 | await asyncio.sleep(10) |
| 4 | return "done" |
| 5 | |
| 6 | async def main(): |
| 7 | try: |
| 8 | result = await asyncio.wait_for(slow(), timeout=2.0) |
| 9 | except asyncio.TimeoutError: |
| 10 | print("timed out!") |
| 11 | |
| 12 | # asyncio.timeout — async context manager (3.11+) |
| 13 | async def main(): |
| 14 | try: |
| 15 | async with asyncio.timeout(2.0): |
| 16 | result = await slow() |
| 17 | except TimeoutError: |
| 18 | print("timed out!") |
| 19 | |
| 20 | # Shield — protect from cancellation |
| 21 | async def main(): |
| 22 | task = asyncio.create_task(slow()) |
| 23 | await asyncio.sleep(0.5) |
| 24 | # task cancelled but shield protects inner operation |
| 25 | result = await asyncio.shield(task) |
| 26 | |
| 27 | # Cancellation handling |
| 28 | async def cancellable(): |
| 29 | try: |
| 30 | await asyncio.sleep(10) |
| 31 | except asyncio.CancelledError: |
| 32 | print("cleaning up before cancellation...") |
| 33 | raise # must re-raise |
| 34 | |
| 35 | # Timeout with asyncio.timeout_at (absolute time) |
| 36 | async def main(): |
| 37 | deadline = asyncio.get_running_loop().time() + 2.0 |
| 38 | async with asyncio.timeout_at(deadline): |
| 39 | result = await slow() |
Debugging & Best Practices
debugging.py
Python
| 1 | # Enable debug mode |
| 2 | # PYTHONASYNCIODEBUG=1 python script.py |
| 3 | # Or: |
| 4 | asyncio.run(main(), debug=True) |
| 5 | |
| 6 | # Debug mode provides: |
| 7 | # - Slow callback warnings (>100ms) |
| 8 | # - Resource warnings (unclosed transports) |
| 9 | # - Coroutine/async gen warnings (never awaited) |
| 10 | # - Stack traces for scheduled callbacks |
| 11 | |
| 12 | # Common mistakes: |
| 13 | |
| 14 | # 1. Forgetting to await |
| 15 | async def fetch(): |
| 16 | return "data" |
| 17 | |
| 18 | result = fetch() # returns coroutine, NOT "data"! |
| 19 | result = await fetch() # correct |
| 20 | |
| 21 | # 2. Blocking the event loop |
| 22 | async def bad(): |
| 23 | time.sleep(1) # blocks whole event loop! |
| 24 | await asyncio.sleep(1) # correct — yields control |
| 25 | |
| 26 | # 3. Mixing asyncio and threading |
| 27 | # Use loop.call_soon_threadsafe() to schedule from threads |
| 28 | |
| 29 | # 4. Not using asyncio.run() properly |
| 30 | # asyncio.run() should be called once per program |
| 31 | |
| 32 | # 5. Unhandled exceptions in tasks |
| 33 | async def main(): |
| 34 | task = asyncio.create_task(fetch()) |
| 35 | await asyncio.sleep(1) |
| 36 | # task might have raised — check task.exception() |
| 37 | if task.done() and task.exception(): |
| 38 | print(f"task error: {task.exception()}") |
| 39 | |
| 40 | # 6. Unawaited coroutines — use create_task for fire-and-forget |
| 41 | async def background(): |
| 42 | while True: |
| 43 | await asyncio.sleep(1) |
| 44 | print("tick") |
| 45 | |
| 46 | async def main(): |
| 47 | task = asyncio.create_task(background()) |
| 48 | await asyncio.sleep(3) |
| 49 | task.cancel() |
$Blueprint — Engineering Documentation·Section ID: PYTHON-ASYNC·Revision: 1.0