Python asyncio Tutorial: Coroutines, Tasks, and Concurrent I/O

Overview

asyncio is Python's standard library for writing concurrent code using a single thread. It is ideal for I/O-bound workloads: network requests, database queries, and file operations. This tutorial covers coroutines, tasks, and the patterns that avoid the most common mistakes.

When to Use asyncio

asyncio does not make CPU-bound code faster. It improves throughput when tasks spend most of their time waiting.

Coroutines and async/await

A coroutine is a function defined with async def. Calling it does not run the body; it returns a coroutine object that must be awaited or scheduled.

import asyncio

async def greet(name):
    await asyncio.sleep(1)
    return f"Hello, {name}"

async def main():
    result = await greet("Alice")
    print(result)

asyncio.run(main())

asyncio.run() creates the event loop, runs the coroutine to completion, and cleans up.

Running Tasks Concurrently

Sequential — Slow

async def main():
    a = await fetch("https://api.example.com/a")   # 1s
    b = await fetch("https://api.example.com/b")   # 1s
    c = await fetch("https://api.example.com/c")   # 1s
    # Total: ~3s

Concurrent — Fast

async def main():
    a, b, c = await asyncio.gather(
        fetch("https://api.example.com/a"),
        fetch("https://api.example.com/b"),
        fetch("https://api.example.com/c"),
    )
    # Total: ~1s

asyncio.gather runs the coroutines concurrently and returns results in order.

Tasks vs Coroutines

async def main():
    task = asyncio.create_task(fetch("https://api.example.com/slow"))
    # Do other work here
    result = await task

Timeouts

async def main():
    try:
        result = await asyncio.wait_for(fetch("https://api.example.com/slow"), timeout=5.0)
    except asyncio.TimeoutError:
        print("Request timed out")

In Python 3.11+, asyncio.timeout() is the preferred context manager form:

async def main():
    async with asyncio.timeout(5.0):
        result = await fetch("https://api.example.com/slow")

Cancellation

async def worker():
    try:
        while True:
            await asyncio.sleep(1)
            print("tick")
    except asyncio.CancelledError:
        print("Cleaning up")
        raise

async def main():
    task = asyncio.create_task(worker())
    await asyncio.sleep(3)
    task.cancel()
    try:
        await task
    except asyncio.CancelledError:
        print("Task cancelled")

Always re-raise CancelledError. Swallowing it breaks cooperative cancellation.

Semaphores for Concurrency Limits

Running 10,000 requests at once overwhelms the target and the client. Bound concurrency with a semaphore.

import asyncio
import aiohttp

async def fetch(session, url, sem):
    async with sem:
        async with session.get(url) as response:
            return await response.text()

async def main(urls):
    sem = asyncio.Semaphore(10)
    async with aiohttp.ClientSession() as session:
        tasks = [fetch(session, url, sem) for url in urls]
        return await asyncio.gather(*tasks)

Producer/Consumer with a Queue

async def producer(queue):
    for i in range(100):
        await queue.put(i)
    await queue.put(None)   # sentinel

async def consumer(queue, name):
    while True:
        item = await queue.get()
        if item is None:
            break
        print(f"{name} processing {item}")
        await asyncio.sleep(0.1)
        queue.task_done()

async def main():
    queue = asyncio.Queue(maxsize=20)
    producers = [asyncio.create_task(producer(queue))]
    consumers = [asyncio.create_task(consumer(queue, f"worker-{i}")) for i in range(3)]
    await asyncio.gather(*producers)
    await queue.join()
    for c in consumers:
        c.cancel()

Blocking Calls

Calling a blocking function inside a coroutine blocks the entire event loop.

# Bad — blocks the loop
async def main():
    time.sleep(5)   # nothing else runs during these 5 seconds

# Good — run in a thread pool
async def main():
    await asyncio.to_thread(time.sleep, 5)

For CPU-bound work, use a process pool instead:

from concurrent.futures import ProcessPoolExecutor

async def main():
    loop = asyncio.get_running_loop()
    with ProcessPoolExecutor() as pool:
        result = await loop.run_in_executor(pool, heavy_computation, data)

Async Context Managers and Iterators

class AsyncResource:
    async def __aenter__(self):
        await self.connect()
        return self

    async def __aexit__(self, exc_type, exc, tb):
        await self.close()

async with AsyncResource() as resource:
    await resource.do_work()

async def paginate(url):
    page = 1
    while True:
        data = await fetch(f"{url}?page={page}")
        if not data:
            break
        for item in data:
            yield item
        page += 1

async for item in paginate("https://api.example.com/items"):
    print(item)

Common Pitfalls

TaskGroup (Python 3.11+)

async def main():
    async with asyncio.TaskGroup() as tg:
        tg.create_task(fetch_and_store("a"))
        tg.create_task(fetch_and_store("b"))
        tg.create_task(fetch_and_store("c"))
    # All tasks completed or an exception propagated

TaskGroup cancels remaining tasks if any task raises, which is safer than bare gather for related work.