Async Python with asyncio

Python’s asyncio library enables concurrent I/O operations on a single thread by using an event loop. Instead of blocking while waiting for a network response or file read, the program suspends and runs other tasks in the meantime.

Coroutines and async def

Traditional synchronous Python blocks the entire thread while waiting for slow operations like network requests or disk reads. Coroutines solve this by allowing a function to pause itself at a specific point — the await expression — and yield control back to the event loop so other work can proceed. This makes it possible to handle many concurrent I/O operations without the complexity and overhead of threads. A coroutine is defined with async def and is not executed immediately — you need to await it or schedule it on the event loop.

import asyncio

async def fetch_user(user_id: int) -> dict:
    """Simulates an async database query."""
    await asyncio.sleep(0.1)   # suspend here — other coroutines can run during this wait
    return {"id": user_id, "name": f"User-{user_id}"}

async def main() -> None:
    user = await fetch_user(42)   # wait for fetch_user to finish, then continue
    print(user)   # {'id': 42, 'name': 'User-42'}

# asyncio.run() creates the event loop, runs the coroutine, then shuts the loop down
asyncio.run(main())

Running Tasks Concurrently with gather

The real power of asyncio comes from running many I/O operations at the same time. asyncio.gather() takes multiple coroutines and runs them concurrently — the total time is determined by the slowest task, not the sum of all tasks. This is a fundamental shift from sequential code: three 1-second requests that would take 3 seconds sequentially take roughly 1 second with gather. Results are returned in the same order as the input coroutines, regardless of completion order.

import asyncio
import time

async def fetch(url: str, delay: float) -> str:
    print(f"  → Fetching {url}")
    await asyncio.sleep(delay)   # simulate network latency without blocking the thread
    print(f"  ← Done {url}")
    return f"Response from {url}"

async def main() -> None:
    start = time.perf_counter()

    # Sequential — takes 1.0 + 1.5 + 0.8 = 3.3 seconds
    # r1 = await fetch("/api/users", 1.0)
    # r2 = await fetch("/api/posts", 1.5)
    # r3 = await fetch("/api/comments", 0.8)

    # Concurrent — takes max(1.0, 1.5, 0.8) ≈ 1.5 seconds
    results = await asyncio.gather(
        fetch("/api/users",    1.0),
        fetch("/api/posts",    1.5),
        fetch("/api/comments", 0.8),
    )

    elapsed = time.perf_counter() - start
    print(f"\nAll done in {elapsed:.2f}s")
    for r in results:
        print(f"  {r}")

asyncio.run(main())

Creating and Managing Tasks

asyncio.gather() is convenient but requires all coroutines to be known upfront. asyncio.create_task() offers more flexibility: it schedules a coroutine on the event loop immediately and returns a Task object you can inspect, cancel, or await later. This is the right tool when you need to fire off background work while continuing to do other things in the same coroutine — for example, starting a periodic health-check while handling a user request.

import asyncio

async def background_job(name: str, seconds: float) -> None:
    print(f"[{name}] Started")
    await asyncio.sleep(seconds)
    print(f"[{name}] Finished after {seconds}s")

async def main() -> None:
    # create_task() schedules both jobs immediately — they start running right away
    task_a = asyncio.create_task(background_job("A", 2.0))
    task_b = asyncio.create_task(background_job("B", 1.0))

    # This line runs while both tasks are progressing in the background
    print("Tasks created, doing other work...")
    await asyncio.sleep(0.5)

    # Await each task to ensure it has finished before we exit main()
    await task_a
    await task_b
    print("All tasks done")

asyncio.run(main())
# Tasks created, doing other work...
# [B] Finished after 1.0s
# [A] Finished after 2.0s
# All tasks done

Real HTTP Concurrency with aiohttp

The standard requests library is synchronous — each call blocks until the response arrives, which prevents any concurrency. aiohttp is the async equivalent: it makes HTTP requests without blocking the event loop, so hundreds of requests can be in-flight at the same time. This makes it the standard choice for web scrapers, API aggregators, and any service that fans out to multiple upstream endpoints. Install with pip install aiohttp.

import asyncio
import aiohttp

async def fetch_json(session: aiohttp.ClientSession, url: str) -> dict:
    # async with keeps the connection alive efficiently within a session
    async with session.get(url, timeout=aiohttp.ClientTimeout(total=5)) as response:
        response.raise_for_status()   # raises an exception for 4xx/5xx responses
        return await response.json()

async def main() -> None:
    urls = [
        "https://jsonplaceholder.typicode.com/posts/1",
        "https://jsonplaceholder.typicode.com/posts/2",
        "https://jsonplaceholder.typicode.com/posts/3",
    ]

    # A single ClientSession reuses the underlying TCP connection pool — more efficient
    async with aiohttp.ClientSession() as session:
        tasks = [fetch_json(session, url) for url in urls]
        # return_exceptions=True prevents one failed request from cancelling the rest
        results = await asyncio.gather(*tasks, return_exceptions=True)

    for url, result in zip(urls, results):
        if isinstance(result, Exception):
            print(f"ERROR {url}: {result}")
        else:
            print(f"OK {url}: title={result['title']!r}")

asyncio.run(main())

Async Context Managers and Iterators

Many async resources — database connections, file handles, HTTP sessions — need setup and teardown that are themselves async operations. Python’s async with statement supports this through __aenter__ and __aexit__ methods, which can await things during enter and exit. This ensures that even cleanup code (closing a connection, flushing a buffer) can yield to the event loop rather than blocking it. Libraries like aiohttp, asyncpg, and aiosqlite all expose their resources this way.

import asyncio

class AsyncDatabaseConnection:
    """Example async context manager showing the __aenter__/__aexit__ protocol."""

    async def __aenter__(self):
        print("Connecting to database...")
        await asyncio.sleep(0.05)   # simulate async connection handshake
        return self   # the value bound by 'as db'

    async def __aexit__(self, exc_type, exc_val, exc_tb):
        print("Closing connection...")
        await asyncio.sleep(0.01)   # simulate async teardown/flush
        # returning None (or False) does not suppress exceptions

    async def query(self, sql: str) -> list[dict]:
        await asyncio.sleep(0.1)    # simulate async query execution
        return [{"id": 1, "name": "row"}]


async def main() -> None:
    # async with guarantees __aexit__ is called even if query() raises
    async with AsyncDatabaseConnection() as db:
        rows = await db.query("SELECT * FROM users")
        print(rows)

asyncio.run(main())

Handling Timeouts and Cancellation

Network calls can hang indefinitely if a server stops responding. Without a timeout, your program silently stalls and never recovers. asyncio.wait_for() wraps any coroutine with a deadline: if the operation doesn’t complete within the given number of seconds, it is cancelled and a TimeoutError is raised. This is essential for any production code that communicates with external services, where you need to fail fast and surface the problem rather than block forever.

import asyncio

async def slow_operation() -> str:
    await asyncio.sleep(10)   # simulates a hung server
    return "done"

async def main() -> None:
    try:
        # Cancel and raise TimeoutError if slow_operation takes more than 2 seconds
        result = await asyncio.wait_for(slow_operation(), timeout=2.0)
    except asyncio.TimeoutError:
        print("Operation timed out!")   # graceful fallback instead of a silent hang

asyncio.run(main())

asyncio Patterns at a Glance