Async Programming with asyncio

Many programs spend most of their time waiting — for web requests, database queries or files. asyncio lets one Python program do other work during those waits, so it can handle hundreds of network operations concurrently on a single thread using async and await.

Coroutines: async def and await

PY basics.py

import asyncio

async def make_tea():
    print("boiling water…")
    await asyncio.sleep(0.2)
    print("tea is ready")
    return "🍵"

async def main():
    drink = await make_tea()
    print("got", drink)

asyncio.run(main())

Output

boiling water…
tea is ready
got 🍵

async def defines a coroutine function. Calling it doesn't run it; it creates a coroutine object.
await runs another coroutine and pauses the current one until it finishes — while paused, other tasks can run.
asyncio.run(main()) starts the event loop and runs the top-level coroutine. Call it once, at your program's entry point.

Note: asyncio.sleep waits without blocking; the ordinary time.sleep would freeze everything.

Running tasks concurrently

Awaiting coroutines one after another runs them in sequence. asyncio.gather runs them concurrently:

PY gather.py

import asyncio
import time

async def fetch(name, seconds):
    await asyncio.sleep(seconds)
    return f"{name} done"

async def main():
    start = time.perf_counter()
    await fetch("a", 0.2)
    await fetch("b", 0.2)
    print(f"sequential: ~{round(time.perf_counter() - start, 1)}s")

    start = time.perf_counter()
    results = await asyncio.gather(fetch("a", 0.2), fetch("b", 0.2), fetch("c", 0.2))
    print(results, f"concurrent: ~{round(time.perf_counter() - start, 1)}s")

asyncio.run(main())

Output

sequential: ~0.4s
['a done', 'b done', 'c done'] concurrent: ~0.2s

Three 0.2-second waits overlap and finish in 0.2 seconds total. With real network requests, this is the difference between a scraper taking minutes or seconds.

Task groups

asyncio.TaskGroup is the modern, safer way to run several tasks: if one fails, the others are cancelled and the errors are reported together:

PY taskgroup.py

import asyncio

async def download(page):
    await asyncio.sleep(0.05 * page)
    return f"page {page}: {page * 100} bytes"

async def main():
    async with asyncio.TaskGroup() as tg:
        tasks = [tg.create_task(download(p)) for p in range(1, 4)]
    for t in tasks:
        print(t.result())

asyncio.run(main())

Output

page 1: 100 bytes
page 2: 200 bytes
page 3: 300 bytes

Timeouts

PY timeout.py

import asyncio

async def slow_query():
    await asyncio.sleep(5)
    return "rows"

async def main():
    try:
        async with asyncio.timeout(0.1):
            await slow_query()
    except TimeoutError:
        print("query took too long — cancelled")

asyncio.run(main())

Output

query took too long — cancelled

Limiting concurrency

Launching thousands of requests at once can overwhelm a server or get you rate-limited. A Semaphore caps how many run at the same time:

PY semaphore.py

import asyncio

active = 0
peak = 0

async def worker(i, limit):
    global active, peak
    async with limit:
        active += 1
        peak = max(peak, active)
        await asyncio.sleep(0.02)
        active -= 1

async def main():
    limit = asyncio.Semaphore(3)
    await asyncio.gather(*(worker(i, limit) for i in range(20)))
    print("max running at once:", peak)

asyncio.run(main())

Output

max running at once: 3

Async iteration and context managers

PY async_for.py

import asyncio

async def ticker(n):
    for i in range(n):
        await asyncio.sleep(0.01)
        yield i

async def main():
    async for tick in ticker(3):
        print("tick", tick)
    squares = [t * t async for t in ticker(4)]
    print(squares)

asyncio.run(main())

Output

tick 0
tick 1
tick 2
[0, 1, 4, 9]

async for loops over async generators — useful for streaming results or paginated APIs. Similarly, async with manages resources that need asynchronous setup and cleanup, like network connections.

Async in practice

To benefit from asyncio, the libraries you call must be async too. Common choices:

Task	Async libraries
HTTP requests	`httpx`, `aiohttp`
Web frameworks	FastAPI, Starlette, Django (async views)
Databases	`asyncpg`, SQLAlchemy's async mode

PY httpx_demo.py

import asyncio
import httpx

async def main():
    urls = [f"https://jsonplaceholder.typicode.com/posts/{i}" for i in range(1, 6)]
    async with httpx.AsyncClient(timeout=10) as client:
        responses = await asyncio.gather(*(client.get(u) for u in urls))
    for r in responses:
        print(r.json()["title"][:40])

asyncio.run(main())

When to use asyncio

Good fit: lots of waiting on I/O — web servers, scrapers, chat bots, API clients calling many endpoints.
Poor fit: CPU-heavy work like image processing or number crunching. Async doesn't make computation faster; use multiprocessing or concurrent.futures.ProcessPoolExecutor for that.
Mixing in blocking code: wrap it with await asyncio.to_thread(func, *args) so it doesn't freeze the event loop.

Try it yourself

import asyncio
import random

random.seed(3)

async def check_service(name):
    delay = random.uniform(0.05, 0.3)
    await asyncio.sleep(delay)
    status = "up" if random.random() > 0.2 else "DOWN"
    return name, status, delay

async def main():
    services = ["auth", "payments", "search", "email", "cdn"]
    results = await asyncio.gather(*(check_service(s) for s in services))
    for name, status, delay in sorted(results, key=lambda r: r[2]):
        print(f"{name:<9} {status:<5} ({delay * 1000:.0f} ms)")

await main()

Console output

Note: the in-browser editor already runs inside an event loop, so the example above uses top-level await main(). In a normal script, call asyncio.run(main()) instead.

Previousitertools & functools NextTesting