Async Programming and asyncio
asyncio is the core standard library module in Python for writing concurrent I/O code using async/await syntax. It is well-suited for network requests, database queries, file I/O, and other wait-heavy tasks, but is not appropriate for CPU-bound tasks (which should use multiprocessing).
Core Concepts
- Coroutine: A function defined with
async def. Calling it returns a coroutine object (does not execute immediately). - Task: Wraps a coroutine into a scheduled unit for concurrent execution.
- Event Loop: Responsible for scheduling all coroutines and tasks.
await: Suspends the current coroutine and yields control to the event loop. The awaited object must be an awaitable (a coroutine, Task, or Future).
Quick Start
import asyncio
async def greet(name: str, delay: float) -> str:
await asyncio.sleep(delay) # Simulate I/O wait (does not block the event loop)
return f"Hello, {name}!"
async def main():
result = await greet("Alice", 1.0)
print(result) # Hello, Alice!
asyncio.run(main()) # Creates an event loop and runs it; Python 3.7+Concurrent Execution: asyncio.gather
gather runs multiple coroutines concurrently and only returns after all of them complete (returning all results):
import asyncio
import time
async def fetch(name: str, delay: float) -> str:
print(f"{name} started")
await asyncio.sleep(delay)
print(f"{name} finished")
return f"{name} result"
async def main():
start = time.perf_counter()
# Three tasks run concurrently; total time ≈ max(1, 2, 1.5) = 2 seconds
results = await asyncio.gather(
fetch("Task A", 1.0),
fetch("Task B", 2.0),
fetch("Task C", 1.5),
)
elapsed = time.perf_counter() - start
print(results) # ['Task A result', 'Task B result', 'Task C result']
print(f"Total time: {elapsed:.2f}s") # ~2.00s
asyncio.run(main())If any coroutine in gather raises an exception, the others are cancelled by default and the exception is re-raised. Set return_exceptions=True to treat exceptions as return values instead.
Tasks (asyncio.create_task)
create_task immediately submits a coroutine to the event loop for scheduling, offering more flexibility than gather:
import asyncio
async def background_job(n: int) -> int:
await asyncio.sleep(n)
return n * n
async def main():
# Submit immediately without waiting for completion
task1 = asyncio.create_task(background_job(2))
task2 = asyncio.create_task(background_job(3))
print("Tasks submitted, doing other work")
await asyncio.sleep(0.1) # Yield control to the event loop
# Wait for tasks to complete and retrieve results
result1 = await task1
result2 = await task2
print(result1, result2) # 4 9
asyncio.run(main())TaskGroup — Structured Concurrency (Python 3.11+)
asyncio.TaskGroup provides a safer concurrency pattern: if any task in the group fails, the others are automatically cancelled:
import asyncio
async def fetch_data(source: str) -> dict:
await asyncio.sleep(0.5)
if source == "bad":
raise ValueError(f"Source {source} is unavailable")
return {"source": source, "data": "..."}
async def main():
async with asyncio.TaskGroup() as tg:
task_a = tg.create_task(fetch_data("api-1"))
task_b = tg.create_task(fetch_data("api-2"))
task_c = tg.create_task(fetch_data("api-3"))
# Results are accessible only after all tasks succeed
print(task_a.result())
print(task_b.result())
print(task_c.result())
asyncio.run(main())Timeout Control
import asyncio
async def slow_op() -> str:
await asyncio.sleep(10)
return "Done"
async def main():
# asyncio.wait_for: cancels the task on timeout and raises TimeoutError
try:
result = await asyncio.wait_for(slow_op(), timeout=1.0)
except asyncio.TimeoutError:
print("Operation timed out!")
# Python 3.11+: recommended to use asyncio.timeout()
try:
async with asyncio.timeout(1.0):
result = await slow_op()
except TimeoutError:
print("Operation timed out!")
asyncio.run(main())Async Context Managers and Iterators
import asyncio
class AsyncDB:
async def __aenter__(self):
print("Connecting to database")
return self
async def __aexit__(self, *args):
print("Closing connection")
async def fetch(self, query: str):
await asyncio.sleep(0.1)
return [{"id": 1}]
async def main():
async with AsyncDB() as db:
rows = await db.fetch("SELECT * FROM users")
print(rows)
# Async iterator
class AsyncStream:
def __init__(self, items: list):
self._items = iter(items)
def __aiter__(self):
return self
async def __anext__(self):
try:
await asyncio.sleep(0.01)
return next(self._items)
except StopIteration:
raise StopAsyncIteration
async def consume():
async for item in AsyncStream([1, 2, 3]):
print(item)
asyncio.run(consume())asyncio.wait — Fine-Grained Control
import asyncio
async def task(n: int) -> int:
await asyncio.sleep(n)
return n
async def main():
tasks = {asyncio.create_task(task(i)) for i in [3, 1, 2]}
# FIRST_COMPLETED: return immediately when the first task finishes
done, pending = await asyncio.wait(tasks, return_when=asyncio.FIRST_COMPLETED)
for t in done:
print(f"Completed: {t.result()}") # 1
for t in pending:
t.cancel() # Cancel remaining tasks
asyncio.run(main())Practical Example: Concurrent HTTP Requests
import asyncio
import aiohttp # pip install aiohttp
async def fetch(session: aiohttp.ClientSession, url: str) -> dict:
async with session.get(url) as resp:
return await resp.json()
async def main():
urls = [
"https://httpbin.org/get?n=1",
"https://httpbin.org/get?n=2",
"https://httpbin.org/get?n=3",
]
async with aiohttp.ClientSession() as session:
results = await asyncio.gather(
*[fetch(session, url) for url in urls]
)
for r in results:
print(r["args"])
asyncio.run(main())asyncio vs threading vs multiprocessing
| Scenario | Recommended Approach |
|---|---|
| Heavy I/O waiting (network requests, databases) | asyncio |
| CPU-bound (numerical computation, compression) | multiprocessing |
| Mixed scenarios or legacy synchronous code | threading |
| Calling a blocking function from async code | loop.run_in_executor() |
import asyncio
from concurrent.futures import ThreadPoolExecutor
def blocking_io() -> str:
import time
time.sleep(1) # Blocking call
return "io done"
async def main():
loop = asyncio.get_event_loop()
# Run the blocking function in a thread pool so it doesn't block the event loop
with ThreadPoolExecutor() as pool:
result = await loop.run_in_executor(pool, blocking_io)
print(result)
asyncio.run(main())