Async I/O with asyncio

Your Python program hangs when waiting for a database query, an HTTP response, or a file read. For I/O-bound tasks like these, threads can help, but they add overhead, synchronization bugs, and memory pressure. The asyncio module gives you a concurrent, single-threaded alternative: you write asynchronous coroutines that pause and resume without blocking the entire program, letting you handle hundreds of I/O operations concurrently in a clean, readable style.

The problem this lesson solves

I/O operations—network requests, file reads, database queries—are orders of magnitude slower than CPU instructions. A single synchronous HTTP call can block your application for hundreds of milliseconds, during which the entire program waits. This is called I/O-bound blocking. Traditional solutions like threading or multiprocessing add complexity: threads share memory and require locks; processes are heavy and don't share state easily.

Imagine a web scraper that fetches 100 pages. A synchronous loop fetches one at a time. With threading, you might use ThreadPoolExecutor—but each thread takes ~8 MB of memory, and you need careful coordination. asyncio solves this by providing a cooperative multitasking model within a single thread.

Core concept / mental model

Think of asyncio as a task scheduler inside your Python process. Instead of many threads, you have an event loop that runs multiple coroutines (async functions). When a coroutine hits an I/O operation, it says: "I'm going to wait—run someone else." The event loop saves its state and executes another ready coroutine. When the I/O finishes, the loop resumes the original coroutine right where it left off.

Key definitions

• Coroutine: An async def function that can be paused and resumed.
• Awaitable: An object you can await (coroutine, asyncio.Task, asyncio.Future).
• Event loop: The engine that schedules and runs coroutines.
• Task: A wrapper around a coroutine that schedules it on the event loop.

🧠 Mental image: Coroutines are like cooperative workers. Each worker picks up a task, begins working, and when it needs to wait (e.g., fetch data from disk), it voluntarily hands control back to the boss (event loop), who assigns another worker. No one is forcibly interrupted.

How it works step by step

1. Define a coroutine: Use async def instead of def.
2. Inside the coroutine, use await to pause execution until an awaitable completes.
3. Run the event loop with asyncio.run(main())—this creates the loop, runs main(), and shuts down the loop.
4. Create multiple tasks using asyncio.create_task() to run coroutines concurrently.
5. Gather results with asyncio.gather(*tasks) to wait for all tasks to finish.

A simple example

import asyncio

async def say_hello(name: str, delay: float):
    await asyncio.sleep(delay)
    print(f"Hello, {name}!")

async def main():
    # Run two coroutines concurrently
    task1 = asyncio.create_task(say_hello("Alice", 2))
    task2 = asyncio.create_task(say_hello("Bob", 1))

    # Wait for both to complete
    await task1
    await task2

asyncio.run(main())

Expected output:

Hello, Bob!
Hello, Alice!

Bob's 1-second delay finishes first, so his message appears before Alice's, even though the coroutines started together. This demonstrates concurrent execution.

The event loop in action

import asyncio

async def count_to_n(name: str, n: int, delay: float):
    for i in range(1, n + 1):
        print(f"{name}: {i}")
        await asyncio.sleep(delay)
    return f"{name} finished"

async def main():
    results = await asyncio.gather(
        count_to_n("A", 3, 0.5),
        count_to_n("B", 2, 0.8)
    )
    print(results)

asyncio.run(main())

Expected output (order may vary slightly):

A: 1
B: 1
A: 2
A: 3
B: 2
['A finished', 'B finished']

Here asyncio.gather runs both coroutines concurrently. The event loop interleaves their execution based on await asyncio.sleep() pauses.

Hands-on walkthrough

Let's build a realistic example: fetch multiple URLs concurrently using aiohttp (an async HTTP library). Install it first:

pip install aiohttp

Then write:

import asyncio
import aiohttp

async def fetch_url(session: aiohttp.ClientSession, url: str) -> str:
    async with session.get(url) as response:
        return await response.text()

async def main():
    urls = [
        "https://httpbin.org/delay/1",
        "https://httpbin.org/delay/2",
        "https://httpbin.org/delay/3"
    ]
    async with aiohttp.ClientSession() as session:
        tasks = [asyncio.create_task(fetch_url(session, url)) for url in urls]
        pages = await asyncio.gather(*tasks)
        print(f"Fetched {len(pages)} pages")

asyncio.run(main())

Expected output:

Fetched 3 pages

All three URLs are fetched concurrently. The total time is roughly the slowest single request (~3 seconds) instead of 6 seconds if done synchronously.

💡 Pro tip: Always use async with for session management—it ensures proper cleanup even if an exception occurs. aiohttp.ClientSession acts as an async context manager.

Compare options / when to choose what

Rule of thumb: Use asyncio when all your I/O libraries support it (the ecosystem is mature: aiohttp, asyncpg, httpx, aioredis, etc.). For blocking libraries, consider threading with run_in_executor.

Troubleshooting & edge cases

Common pitfalls

1. Forgetting await: Calling an async function without await returns a coroutine object, not the result. Always await async calls.

python result = fetch_url(session, url) # ❌ Returns coroutine, not data result = await fetch_url(session, url) # ✅ Correct

1. Mixing sync and async: You cannot await a synchronous function. If you need to run blocking code in an async context, use asyncio.to_thread() (Python 3.9+).

```python import time

async def main(): # This will block the event loop! time.sleep(1) # ❌ # Use this instead: await asyncio.to_thread(time.sleep, 1) # ✅ ```

1. asyncio.run() only once: In a script, call asyncio.run() once. Nesting asyncio.run() inside a coroutine raises a RuntimeError. Use asyncio.get_event_loop() if needed for advanced use cases.

Error handling

import asyncio

async def failing_coro():
    raise ValueError("Something went wrong")

async def main():
    try:
        await failing_coro()
    except ValueError as e:
        print(f"Caught: {e}")

asyncio.run(main())

Expected output:

Caught: Something went wrong

When using asyncio.gather(), you can use return_exceptions=True to collect exceptions as results instead of propagating them.

async def main():
    results = await asyncio.gather(
        failing_coro(),
        asyncio.sleep(1),
        return_exceptions=True
    )
    print(results)  # [ValueError('Something went wrong'), None]

What you learned & what's next

You now understand: - The problem asyncio solves: I/O blocking in single-threaded code. - The mental model: cooperative multitasking with an event loop. - The step-by-step mechanics: coroutines, await, tasks, and gather. - The hands-on application: fetching URLs concurrently. - When to choose asyncio over threading or multiprocessing. - Common troubles like missing await or mixing sync/async.

Next, you'll learn how to integrate async code with synchronous libraries using asyncio.to_thread and run_in_executor, enabling you to unify blocking and non-blocking code in one application.