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())

Last updated on 2026-06-23

Coroutines