asyncio

Updated

September 8, 2026

asyncio

async def marks a coroutine function. Calling it does not run the body; it builds a coroutine object. await pauses this coroutine until that one finishes. asyncio.run is the boring entry point: it starts the event loop, runs one coroutine to completion, and closes the loop. Use asyncio when you have many I/O waits in one process and you want to overlap them without a thread per wait.

Mental model

The loop runs one coroutine until it awaits. Then it can run another. asyncio.sleep is a fake I/O wait. Real I/O uses libraries that actually yield (sockets, subprocess, some HTTP clients). time.sleep inside a coroutine blocks the loop. Do not do that.

asyncio.TaskGroup starts child tasks and waits for all of them. If one fails, the group cancels the others and raises an ExceptionGroup.

asyncio.timeout cancels the block when the budget is gone. It raises TimeoutError. There is no infinite while True in this chapter.

Worked examples

Case 1: async def, await, asyncio.run

Save as async_fire.py. await asyncio.sleep yields the loop. Sequential awaits still add, because you waited for each ticket before starting the next.

# async_fire.py
import asyncio


async def fire(ticket):
    await asyncio.sleep(0.05)
    return f"fired {ticket}"


async def main():
    for ticket in ["T-11", "T-12"]:
        print(await fire(ticket))


if __name__ == "__main__":
    asyncio.run(main())

Run:

uv run python async_fire.py

Output:

fired T-11
fired T-12

asyncio.run(main()) belongs under the main guard, like any other script entry. Do not call asyncio.run twice in one process if you can avoid it.

Case 2: TaskGroup overlaps waits

Save as async_shifts.py. Both shifts sleep at the same time. The group does not exit until both tasks finish. .result() is safe after the async with block.

# async_shifts.py
import asyncio
import time


async def load_shift(name, delay):
    await asyncio.sleep(delay)
    return f"{name} ready"


async def main():
    started = time.perf_counter()
    async with asyncio.TaskGroup() as tg:
        am = tg.create_task(load_shift("am", 0.05))
        pm = tg.create_task(load_shift("pm", 0.05))
    elapsed = time.perf_counter() - started
    print(am.result())
    print(pm.result())
    print(f"elapsed < 0.10: {elapsed < 0.10}")


if __name__ == "__main__":
    asyncio.run(main())

Run:

uv run python async_shifts.py

Output:

am ready
pm ready
elapsed < 0.10: True

asyncio.gather is the older pile of awaitables. TaskGroup is the 3.11+ default: structured, cancels siblings on failure.

Case 3: timeout

Save as async_timeout.py. The grill takes too long. The timeout cancels the wait. The program still exits.

# async_timeout.py
import asyncio


async def grill(ticket):
    await asyncio.sleep(1)
    return f"done {ticket}"


async def main():
    try:
        async with asyncio.timeout(0.05):
            print(await grill("T-11"))
    except TimeoutError:
        print("grill too slow")


if __name__ == "__main__":
    asyncio.run(main())

Run:

uv run python async_timeout.py

Output:

grill too slow

TimeoutError here is the built-in exception. Bound every external wait that can hang. A kitchen that never answers should not pin your process.

Case 4: Many tickets, one group

Save as async_rail.py. Create a task per ticket inside the group. Print in a stable order after the group completes.

# async_rail.py
import asyncio


async def fire(ticket):
    await asyncio.sleep(0.01)
    return f"fired {ticket}"


async def main():
    tickets = ["T-11", "T-12", "T-13"]
    async with asyncio.TaskGroup() as tg:
        tasks = [tg.create_task(fire(t)) for t in tickets]
    for task in tasks:
        print(task.result())


if __name__ == "__main__":
    asyncio.run(main())

Run:

uv run python async_rail.py

Output:

fired T-11
fired T-12
fired T-13

The tasks may finish in any order. You print from the list you created, so the report is stable.

The trap

Save as async_sleep_wrong.py. time.sleep blocks the thread the loop runs on. The second ticket cannot start until the first sleep is over. You wrote async and got sequential blocking I/O with extra syntax.

# async_sleep_wrong.py
import asyncio
import time


async def fire(ticket):
    time.sleep(0.05)
    return f"fired {ticket}"


async def main():
    started = time.perf_counter()
    async with asyncio.TaskGroup() as tg:
        tg.create_task(fire("T-11"))
        tg.create_task(fire("T-12"))
    elapsed = time.perf_counter() - started
    print(f"elapsed >= 0.10: {elapsed >= 0.10}")


if __name__ == "__main__":
    asyncio.run(main())

Run:

uv run python async_sleep_wrong.py

Output:

elapsed >= 0.10: True

Replace time.sleep with await asyncio.sleep and the elapsed check flips. async def is not a speed potion. Only await points overlap.

The boring rule

  • asyncio.run(main()) once, under the main guard.
  • await real I/O (or asyncio.sleep in examples). Never time.sleep in a coroutine.
  • TaskGroup for a known set of child tasks. Do not fire-and-forget.
  • asyncio.timeout around waits that can hang. The process must still exit.
  • Sequential await in a for is still sequential. Overlap needs tasks.

Try this

  1. In async_fire.py, create both fire tasks in a TaskGroup and print .result() after the block.
  2. Change Case 3’s timeout to 2 seconds so the grill succeeds and prints done T-11.
  3. Raise RuntimeError("burnt") in one load_shift. Catch ExceptionGroup around the TaskGroup.
  4. Add a third shift late to Case 2. Keep the elapsed check under 0.10.