Async Programming with asyncio

Most programs spend time waiting — for the network, a database, a disk. asyncio lets one thread juggle thousands of waiting tasks: while one task awaits a response, others run.

Define coroutines with async def, pause them with await, and start the event loop with asyncio.run(main()). Run tasks concurrently with asyncio.gather or, better, asyncio.TaskGroup (Python 3.11+), which cancels the other tasks if one fails.

Limit concurrency with asyncio.Semaphore (so you don't hit an API with 1,000 requests at once) and bound waiting with asyncio.timeout. Never call blocking functions like time.sleep or requests.get inside async code — use async libraries (such as httpx) or asyncio.to_thread.

async_demo.pyPython

import asyncio
import random
import time


async def fetch_result(student_id: int, limit: asyncio.Semaphore) -> tuple[int, int]:
    async with limit:                              # at most N at a time
        await asyncio.sleep(random.uniform(0.1, 0.3))   # pretend network call
        return student_id, random.randint(20, 100)


async def main() -> None:
    limit = asyncio.Semaphore(10)
    start = time.perf_counter()

    async with asyncio.TaskGroup() as tg:
        tasks = [tg.create_task(fetch_result(i, limit)) for i in range(50)]

    results = [t.result() for t in tasks]
    print(f"{len(results)} results in {time.perf_counter() - start:.2f}s")  # ~1s, not ~10s

    try:
        async with asyncio.timeout(0.05):
            await asyncio.sleep(1)
    except TimeoutError:
        print("timed out as expected")

    checksum = await asyncio.to_thread(sum, range(10_000_000))  # blocking work off the loop
    print(checksum)


asyncio.run(main())

Key points

asyncio shines for I/O-bound work: many requests, sockets, DB calls.
Prefer TaskGroup; limit concurrency with a Semaphore.
Never block the event loop — use async libraries or asyncio.to_thread.

Exercise

Install httpx and fetch users 1–10 from https://jsonplaceholder.typicode.com/users/{id} concurrently with httpx.AsyncClient and a TaskGroup, limited to 3 at a time. Compare the time against fetching them one by one.

Show solution

Try the exercise yourself first — then compare your approach with this one.

All ten requests start together inside a TaskGroup, but the Semaphore(3) lets only three run at a time. The sequential version awaits each request before starting the next, so it takes roughly ten times as long as one request.

TerminalShell

pip install httpx

users_async.pyPython

import asyncio
import time

import httpx

BASE_URL = "https://jsonplaceholder.typicode.com"


async def fetch_user(client: httpx.AsyncClient, user_id: int, limit: asyncio.Semaphore) -> str:
    async with limit:
        response = await client.get(f"{BASE_URL}/users/{user_id}", timeout=5)
        response.raise_for_status()
        return response.json()["name"]


async def concurrent() -> list[str]:
    limit = asyncio.Semaphore(3)
    async with httpx.AsyncClient() as client, asyncio.TaskGroup() as tg:
        tasks = [tg.create_task(fetch_user(client, i, limit)) for i in range(1, 11)]
    return [t.result() for t in tasks]


async def sequential() -> list[str]:
    limit = asyncio.Semaphore(1)
    async with httpx.AsyncClient() as client:
        return [await fetch_user(client, i, limit) for i in range(1, 11)]


async def main() -> None:
    for label, run in [("sequential", sequential), ("concurrent (3 at a time)", concurrent)]:
        start = time.perf_counter()
        names = await run()
        print(f"{label:<26} {time.perf_counter() - start:.2f}s  {names[:3]}...")


asyncio.run(main())

Key points

Exercise

Check your understanding