Async Python with asyncio

In the Python for Beginners series every program you have written so far has run top to bottom, one line after the next. Chapter 22 of this 24-chapter course adds the exception: asyncio, which lets a single thread interleave many tasks — the trick behind Python servers that hold thousands of connections open at once.

From synchronous to asynchronous

Synchronous code is a queue: line two waits for line one. That is exactly right for CPU work — crunching numbers, parsing, rendering — and wrong for the other common operation, waiting. A network program spends most of its life blocked on a socket or a timer, and a blocked thread does nothing at all. Asynchronous code keeps one thread but switches between tasks at well-defined points, so waiting time is spent running someone else’s work. The switcher is the event loop: it runs one coroutine until that coroutine hits an await, parks it, and picks the next ready task from the queue.

Coroutines: async def, await and asyncio.run

An async def function is a coroutine. Calling it does not run it — it creates an awaitable object. await pauses that coroutine until the inner operation finishes, and while it waits the event loop is free to run other tasks. asyncio.run() is the only entry point you need: it creates the loop, runs your coroutine to completion, and shuts everything down. Below, gather schedules both coroutines at once — notice that two finishes first even though it started second:

import asyncio

async def one():
    print("one started")
    await asyncio.sleep(0.3)
    print("one finished")

async def two():
    print("two started")
    await asyncio.sleep(0.1)
    print("two finished")

async def main():
    await asyncio.gather(one(), two())

asyncio.run(main())

Expected output:

one started
two started
two finished
one finished

Had we awaited one() and then two() in sequence, the output would be strictly ordered — four lines, one after the other. Also note what asyncio.sleep is not: time.sleep blocks the whole thread, while asyncio.sleep merely reschedules the current task. That is the entire difference between blocking and yielding.

Running tasks together: gather, create_task and TaskGroup

Three tools cover almost every need. asyncio.gather(*coros) starts several coroutines and waits for all their results. asyncio.create_task(coro) starts one immediately and hands back a handle you can await later. And asyncio.TaskGroup (Python 3.11 and newer) is the structured form: it waits for every child and raises if any of them fails, so partial failures cannot slip through:

import asyncio

async def fetch(source, delay):
    print(f"start {source}")
    await asyncio.sleep(delay)
    print(f"done {source}")
    return f"<{source}>"

async def main():
    async with asyncio.TaskGroup() as tg:
        first = tg.create_task(fetch("alpha", 0.1))
        second = tg.create_task(fetch("beta", 0.1))
    print(first.result(), second.result())

asyncio.run(main())

Expected output:

start alpha
start beta
done alpha
done beta
<alpha> <beta>

When async helps — and when it does not

Async pays for itself wherever tasks spend their time waiting: HTTP calls, database queries, websockets, asyncio.sleep standing in for a slow dependency. It buys nothing for CPU-bound work — the loop can only switch at an await, so a function that computes for ten seconds without awaiting blocks every other task behind it. Move that work to a thread with asyncio.to_thread, into a process pool, or keep it plain synchronous.

Async also reaches iteration. An async def generator with yield is consumed with async for: async for tick in countdown(3): print("tick", tick) prints tick 0, tick 1 and tick 2, each line appearing only after the awaited delay. You will meet the same shape in JavaScript’s async / await and event loop chapters — one mental model, two languages.

Worked example: three sources at once

The payoff. Three “fetches”, each waiting 0.3 seconds like a network call would; create_task starts them all immediately and gather collects the results. Thanks to asyncio.sleep the batch finishes in well under the 0.9 seconds three sequential calls would need — and it runs offline:

import asyncio
import time

async def fetch(source, delay):
    print(f"start {source}")
    await asyncio.sleep(delay)
    print(f"done {source}")
    return f"<{source}>"

async def main():
    start = time.perf_counter()
    tasks = [asyncio.create_task(fetch(s, 0.3)) for s in ("docs", "blog", "api")]
    pages = await asyncio.gather(*tasks)
    elapsed = time.perf_counter() - start
    print("".join(pages))
    print(f"finished well under 0.9s: {elapsed < 0.9}")

asyncio.run(main())

Expected output:

start docs
start blog
start api
done docs
done blog
done api
<docs><blog><api>
finished well under 0.9s: True

All three start lines print before any done line — that interleaving, not the clock, is the proof that the tasks overlapped. Sequential awaits would have printed six strictly alternating pairs instead.

Common mistakes

• Forgetting async or await — calling a coroutine without await returns an object you never run (Python warns about it), and awaiting a plain function is a syntax error. Your first test catches both.
• Blocking inside a coroutine — time.sleep, heavy loops and synchronous network calls freeze the whole loop. Use asyncio.sleep, and move CPU work to asyncio.to_thread.
• Running async code the wrong way — asyncio.run() belongs in ordinary synchronous code, once, at the top level — never inside a coroutine, where you would use await.
• Firing tasks you never await — create a task and forget the handle, and its exception may vanish or the program may exit before the work finishes. Keep the tasks and gather them.
• Reaching for async everywhere — a script that reads one file and prints a line is clearer synchronous. Reach for the loop only when you are genuinely waiting on many things at once.

Practice checklist

1. Run the interleaving example, then rewrite it with two sequential await calls and compare the output.
2. Add a third task to the TaskGroup example and confirm its start and done lines interleave with the others.
3. Measure sequential versus concurrent timing with time.perf_counter() around five 0.2-second fetches.
4. Write an async def generator that yields the squares of 1–4 with await asyncio.sleep(0.1) and consume it with async for.
5. Make one fetch inside a TaskGroup raise an exception and read the error the group reports.

What's next?

You can now annotate, test and schedule your code. Next you stop reinventing wheels: Standard Library Essentials tours the batteries that ship with Python itself.