Async/Await in Python: A Practical Guide

Asynchronous programming allows a program to handle multiple operations concurrently without creating multiple threads or processes. Python’s async/await syntax, introduced in Python 3.5, provides a clean way to write concurrent code using coroutines. This is especially useful for I/O-bound tasks like web requests, database queries, and file operations, where the program would otherwise spend most of its time waiting.

Understanding the Event Loop

The event loop is the core of Python’s async system. It runs a single thread, continually checking for tasks that are ready to execute. When a coroutine encounters an await expression, it yields control back to the event loop, which can then run another coroutine while waiting for the I/O operation to complete. The asyncio module provides the event loop, and in modern Python (3.10+), asyncio.run() handles loop creation and cleanup automatically.

import asyncio

async def fetch_data(url):
    print(f"Fetching {url}...")
    await asyncio.sleep(1)  # Simulate network delay
    return f"Data from {url}"

async def main():
    # Run multiple tasks concurrently
    tasks = [
        fetch_data("https://api.example.com/users"),
        fetch_data("https://api.example.com/posts"),
        fetch_data("https://api.example.com/comments"),
    ]
    results = await asyncio.gather(*tasks)
    for r in results:
        print(r)

asyncio.run(main())

Async vs Synchronous Performance

The real benefit of async becomes apparent with many I/O operations. A synchronous version of the above would take 3 seconds (one after another), while the async version completes in about 1 second because all three requests run concurrently. This scales linearly – fetching 100 URLs synchronously takes 100 seconds; asynchronously, it still takes about 1 second (limited by bandwidth and server capacity). The sweet spot for async is high-latency, I/O-bound workloads with hundreds or thousands of concurrent operations.

Common Pitfalls

Blocking the event loop is the most common mistake. Calling time.sleep(), requests.get(), or any synchronous blocking function inside an async function blocks the entire event loop, defeating the purpose of async. Always use asyncio.sleep() instead of time.sleep(), and use async HTTP libraries like aiohttp or httpx instead of requests. Another pitfall is forgetting to await a coroutine – this returns a coroutine object instead of executing it, which can lead to silent bugs because the coroutine is never scheduled.

import aiohttp

async def fetch_json(session, url):
    async with session.get(url) as resp:
        return await resp.json()

async def fetch_all(urls):
    async with aiohttp.ClientSession() as session:
        tasks = [fetch_json(session, url) for url in urls]
        return await asyncio.gather(*tasks)

urls = [f"https://api.example.com/page/{i}" for i in range(50)]
results = asyncio.run(fetch_all(urls))
print(f"Fetched {len(results)} pages")

Python 3.11+ includes high-level task groups (asyncio.TaskGroup) for structured concurrency, making error handling more predictable. When any task in a group fails, all sibling tasks are cancelled automatically, preventing orphaned background tasks.

Real-World Async Patterns

In production applications, you will often combine asyncio with other concurrency patterns. A common pattern is the producer-consumer setup where one coroutine fetches data from an API and another processes it. Using asyncio.Queue, you can coordinate work between coroutines with backpressure—if the consumer is slower than the producer, the queue fills up and the producer waits. Another pattern is using asyncio.timeout() (Python 3.11+) to set a maximum wait time for operations, preventing a single slow request from holding up the entire pipeline. For CPU-bound tasks within an async application, use loop.run_in_executor() with ThreadPoolExecutor to offload work to a thread pool without blocking the event loop.

async def worker(name, queue):
    while True:
        item = await queue.get()
        print(f"Worker {name}: processing {item}")
        await asyncio.sleep(0.2)
        queue.task_done()

async def main():
    queue = asyncio.Queue()
    workers = [asyncio.create_task(worker(f"W{i}", queue)) for i in range(3)]
    for i in range(20):
        await queue.put(f"task-{i}")
    await queue.join()
    for w in workers:
        w.cancel()

asyncio.run(main())

Structured Concurrency with TaskGroups

Python 3.11 introduced asyncio.TaskGroup for structured concurrency. TaskGroup ensures that all child tasks complete before the group exits, and if any task raises an exception, all sibling tasks are cancelled. This prevents orphaned tasks continuing after an error. The ExceptionGroup collects multiple exceptions raised concurrently. Structured concurrency makes async code more predictable—the lifetime of tasks is bounded by the scope of the TaskGroup. For new async code targeting Python 3.11+, prefer TaskGroup over asyncio.gather() for better error handling and resource management.

Async Context Managers and Async Iterators

Python’s async context managers (async with) and async iterators (async for) extend the async paradigm to resource management. Async context managers, defined with __aenter__ and __aexit__, handle async resource setup and teardown—essential for database connections, HTTP sessions, and file handles. The aiofiles library provides async file operations, and aiohttp.ClientSession is an async context manager that properly closes connections. Async iterators (__aiter__ and __anext__) enable paginated API consumption where each page is fetched asynchronously: async for page in api.paginate(): processes results without blocking. The async generator syntax (async def gen(): yield item) creates async iterators with cleaner code. Python 3.10+ supports asynchronous iteration in list comprehensions: [x async for x in async_gen()]. Standard library modules like contextlib provide @asynccontextmanager decorator for simple async context managers.