Python asyncio Tutorial

Python’s asyncio module lets you write concurrent code using async/await syntax. It excels at I/O-bound workloads — HTTP requests, database queries, file operations — where you spend most of your time waiting.

Your first coroutine

import asyncio

async def say_hello():
    print("Hello")
    await asyncio.sleep(1)
    print("World")

asyncio.run(say_hello())

async def defines a coroutine. await suspends execution until the awaited operation completes. asyncio.run() starts the event loop.

Running coroutines concurrently

async def fetch(name, delay):
    print(f"Fetching {name}...")
    await asyncio.sleep(delay)
    print(f"Done: {name}")
    return f"{name} data"

async def main():
    results = await asyncio.gather(
        fetch("users", 2),
        fetch("posts", 1),
        fetch("comments", 3),
    )
    print(results)

asyncio.run(main())

asyncio.gather() runs all coroutines concurrently. Total time is 3 seconds (the longest), not 6 (the sum).

Tasks

Create a task to start a coroutine running in the background.

async def main():
    task1 = asyncio.create_task(fetch("users", 2))
    task2 = asyncio.create_task(fetch("posts", 1))

    # Do other work while tasks run...
    await asyncio.sleep(0.5)
    print("Tasks are running in the background")

    result1 = await task1
    result2 = await task2
    print(result1, result2)

Real HTTP requests with aiohttp

import aiohttp

async def fetch_url(session, url):
    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/1",
    ]

    async with aiohttp.ClientSession() as session:
        tasks = [fetch_url(session, url) for url in urls]
        results = await asyncio.gather(*tasks)
        print(f"Fetched {len(results)} pages")

asyncio.run(main())

Semaphores for rate limiting

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

async def main():
    sem = asyncio.Semaphore(5)  # max 5 concurrent requests

    async with aiohttp.ClientSession() as session:
        tasks = [
            fetch_with_limit(sem, session, f"https://httpbin.org/get?id={i}")
            for i in range(50)
        ]
        results = await asyncio.gather(*tasks)
        print(f"Completed {len(results)} requests")

Timeouts

async def slow_operation():
    await asyncio.sleep(10)
    return "done"

async def main():
    try:
        result = await asyncio.wait_for(slow_operation(), timeout=3.0)
    except asyncio.TimeoutError:
        print("Operation timed out!")

Async generators

async def count_up(limit):
    for i in range(limit):
        await asyncio.sleep(0.1)
        yield i

async def main():
    async for number in count_up(10):
        print(number)

Async context managers

class DatabaseConnection:
    async def __aenter__(self):
        print("Connecting to database...")
        await asyncio.sleep(0.5)
        return self

    async def __aexit__(self, exc_type, exc_val, exc_tb):
        print("Closing connection...")
        await asyncio.sleep(0.1)

    async def query(self, sql):
        await asyncio.sleep(0.2)
        return [{"id": 1, "name": "Alice"}]

async def main():
    async with DatabaseConnection() as db:
        results = await db.query("SELECT * FROM users")
        print(results)

Error handling

async def risky_operation(n):
    if n == 3:
        raise ValueError(f"Bad value: {n}")
    await asyncio.sleep(0.1)
    return n * 2

async def main():
    results = await asyncio.gather(
        risky_operation(1),
        risky_operation(2),
        risky_operation(3),
        return_exceptions=True,
    )

    for result in results:
        if isinstance(result, Exception):
            print(f"Error: {result}")
        else:
            print(f"Result: {result}")

return_exceptions=True prevents one failure from cancelling everything.

Queues

async def producer(queue):
    for i in range(10):
        await queue.put(i)
        print(f"Produced: {i}")
    await queue.put(None)  # sentinel

async def consumer(queue):
    while True:
        item = await queue.get()
        if item is None:
            break
        print(f"Consumed: {item}")
        await asyncio.sleep(0.3)

async def main():
    queue = asyncio.Queue(maxsize=3)
    await asyncio.gather(producer(queue), consumer(queue))

Summary

asyncio lets Python handle thousands of concurrent I/O operations on a single thread. Use async/await for coroutines, gather() for parallel execution, semaphores for rate limiting, and queues for producer-consumer patterns. It is the right tool for I/O-bound workloads — not CPU-bound ones.