The async/await Mental Model: What Asynchronous Python Actually Does

The most common mistake developers make with Python's async/await is thinking it's threading. It isn't. async/await is cooperative scheduling: a single thread runs an event loop that hands control to one coroutine at a time, and each coroutine voluntarily yields control back at await points so the loop can run someone else. Nothing preempts a coroutine; if you write a CPU-bound loop without any await, every other coroutine starves until you finish. The good news: if you build the right mental model up front, async Python stops being mysterious. This entry walks through the model in steps. What the event loop is, what async def creates, what await actually does, and how to combine coroutines with the modern patterns (asyncio.TaskGroup, asyncio.gather, asyncio.to_thread) that make production async code maintainable. Part of our Modern Python 2026 guide.

Why Async Exists

Most production Python is I/O-bound: it spends most of its time waiting on network calls, database queries, file reads, message queues. Each wait blocks the calling thread, which sits idle doing nothing. The classic fix is threading: spawn one thread per concurrent operation. Threading works but costs memory per thread (typically 1-8 MB for the stack), incurs OS context-switch overhead, and hits hard limits in the thousands of threads on most systems.

async/await offers a different deal. A single thread runs an event loop that juggles many concurrent operations. When one operation hits I/O, it yields to the loop, which immediately runs the next ready operation. The result: a single Python process can manage tens of thousands of concurrent connections with modest memory.

The Event Loop

The event loop is a simple piece of machinery: it maintains a queue of "ready" tasks (coroutines that can run right now) and a registry of "waiting" tasks (coroutines paused on I/O). It repeats one rule: pick a ready task, run it until it awaits, then put it in the waiting set or back in the queue.

# Conceptual event loop in plain Python (not the real thing):
def run_loop(tasks):
    ready = list(tasks)
    while ready:
        task = ready.pop(0)
        try:
            io = task.send(None)        # Run until next await
        except StopIteration:
            continue                    # Task completed
        # Schedule resume when the I/O finishes:
        io.callback = lambda: ready.append(task)

The real asyncio event loop is more elaborate (it integrates with epoll, kqueue, IOCP, etc.), but the mental model is exactly this: pop a task, run it to its next await, repeat.

async def Creates a Coroutine Object

The biggest stumbling block for beginners: calling an async def function doesn't run it. It returns a coroutine object, an object representing pending work that needs the event loop to drive it.

async def greet(name: str) -> str:
    return f"Hello {name}"

coro = greet("Alice")
print(coro)
# <coroutine object greet at 0x...>

# The function body hasn't run yet!
# We need the event loop to drive it:
import asyncio
result = asyncio.run(coro)
print(result)
# 'Hello Alice'

This is what PEP 492 introduced in Python 3.5. A coroutine is a paused state-machine; the event loop drives it by calling .send() repeatedly until it raises StopIteration.

await Is a Yield Point

Inside an async function, await on a coroutine or future does two things: it tells the event loop "I'm waiting on this; come back when it's ready", and it pauses the current coroutine at that point. The loop is free to run other ready coroutines until the awaited operation completes.

import asyncio

async def fetch_user(user_id: int) -> dict:
    print(f"start fetch_user({user_id})")
    await asyncio.sleep(1)          # YIELD POINT: loop runs others for 1s
    print(f"finish fetch_user({user_id})")
    return {"id": user_id, "name": f"User{user_id}"}

async def main():
    # These run CONCURRENTLY because await asyncio.sleep yields:
    users = await asyncio.gather(
        fetch_user(1),
        fetch_user(2),
        fetch_user(3),
    )
    print(users)

asyncio.run(main())
# start fetch_user(1)
# start fetch_user(2)
# start fetch_user(3)
# (1 second passes during which all three are awaiting)
# finish fetch_user(1)
# finish fetch_user(2)
# finish fetch_user(3)
# [{'id': 1, ...}, {'id': 2, ...}, {'id': 3, ...}]

If asyncio.sleep(1) blocked the thread, the three would take 3 seconds total. Because it yields to the loop, they take 1 second.

asyncio.run() Is the Entry Point

asyncio.run(coro) creates a new event loop, runs the coroutine to completion, then closes the loop. It's the canonical entry point for most async programs.

import asyncio

async def main():
    print("hello")
    await asyncio.sleep(1)
    print("world")

asyncio.run(main())

One asyncio.run() call per program, typically at the bottom of main.py. Inside the main coroutine you spawn tasks, gather results, and the loop drives everything until main() returns.

Tasks vs Coroutines

A coroutine is a paused object; a task is a coroutine wrapped by the loop for active scheduling. You create a task with asyncio.create_task(coro); the task starts running on the next loop tick, even before you await it.

async def background_work():
    await asyncio.sleep(2)
    print("background done")

async def main():
    # Schedule the task; loop picks it up immediately
    task = asyncio.create_task(background_work())

    # Do other work meanwhile
    print("main doing other work")
    await asyncio.sleep(1)
    print("main partway")

    # Wait for the task to finish
    await task
    print("main finished")

Output:

main doing other work
(1 second passes)
main partway
(1 more second passes; background_work finishes)
background done
main finished

Without create_task, the coroutine never runs. The common bug is to write background_work() alone, expecting it to start; nothing happens because no task wraps it.

asyncio.TaskGroup: Structured Concurrency (3.11+)

Modern asyncio (Python 3.11+) introduced asyncio.TaskGroup: a context manager that ensures all tasks created within it complete (or are cancelled) before the block exits, and propagates exceptions correctly via PEP 654 ExceptionGroups.

import asyncio

async def fetch_user(user_id: int) -> dict:
    await asyncio.sleep(0.1)
    if user_id == 0:
        raise ValueError(f"bad user_id: {user_id}")
    return {"id": user_id}

async def main():
    try:
        async with asyncio.TaskGroup() as tg:
            t1 = tg.create_task(fetch_user(1))
            t2 = tg.create_task(fetch_user(2))
            t3 = tg.create_task(fetch_user(0))   # Will fail
    except* ValueError as eg:
        # PEP 654 except* catches groups of exceptions
        for err in eg.exceptions:
            print(f"caught: {err}")

asyncio.run(main())
# caught: bad user_id: 0

TaskGroup gives structured concurrency: if one task raises, the others are cancelled cleanly. The except* syntax catches ExceptionGroups, which TaskGroup uses to bundle multiple failures.

asyncio.gather: The Older Pattern

Before TaskGroup, asyncio.gather() was the canonical way to run several coroutines concurrently and collect results. It still works and is simpler for the common "run N coroutines, get N results" case.

async def main():
    results = await asyncio.gather(
        fetch_user(1),
        fetch_user(2),
        fetch_user(3),
    )
    # results is a list of three dicts in input order

asyncio.run(main())

For new code on 3.11+, prefer TaskGroup when you need exception safety. Use gather for simple cases or when targeting older Python.

The Blocking Pitfall

The single biggest mistake in async Python: calling a blocking function inside an async function. The event loop has no way to interrupt synchronous code; if you call time.sleep(5) from a coroutine, the entire loop freezes for 5 seconds and no other coroutine runs.

import asyncio
import time

async def bad():
    # WRONG: time.sleep blocks the loop
    time.sleep(2)
    return "done"

async def good():
    # RIGHT: asyncio.sleep yields to the loop
    await asyncio.sleep(2)
    return "done"

Sync libraries (the older requests, most database drivers, file I/O) all block. To use them from async code, dispatch to a thread:

import asyncio
import requests

async def fetch_url(url: str) -> str:
    # Move the blocking call to a thread pool
    response = await asyncio.to_thread(requests.get, url)
    return response.text

asyncio.to_thread (Python 3.9+) runs the given function in a thread pool and gives you a coroutine that awaits its result. The event loop stays responsive; the blocking work happens out of the way.

async Iterators and Context Managers

The async equivalents of regular iterators and context managers exist. async for iterates over an async iterator; async with uses an async context manager.

async def consume_stream():
    async for chunk in stream:               # __aiter__ / __anext__
        await process(chunk)

async def use_resource():
    async with aiofiles.open("data.txt") as f:    # __aenter__ / __aexit__
        contents = await f.read()
        await process(contents)

The async dunder pair (__aiter__/__anext__, __aenter__/__aexit__) is covered briefly in the dunders guide. They follow the same pattern as the sync versions, but the methods are coroutines.

Async Generators

You can combine async def with yield to make an async generator: a coroutine that produces values one at a time, with await points between them. Used for streaming responses, polling queues, or any data source that arrives over time.

import asyncio

async def stream_events(client) -> "AsyncIterator[dict]":
    while True:
        event = await client.poll()       # yields control while polling
        if event is None:
            await asyncio.sleep(1)
            continue
        yield event                        # produces a value, pauses

async def main():
    async for event in stream_events(client):
        await process(event)

asyncio.run(main())

The async for loop drives the generator: each iteration awaits the next yielded value, which can take arbitrary time without blocking the loop. Async generators are the right tool for paginated APIs, server-sent events, message queue consumers, and any I/O source that produces an open-ended sequence.

asyncio.Queue for Producer-Consumer

For decoupling producers and consumers in async code, asyncio.Queue is the standard tool. Producers await queue.put(item); consumers await queue.get(); both yield to the loop when the queue is full or empty.

import asyncio

async def producer(queue: asyncio.Queue, urls: list[str]) -> None:
    for url in urls:
        await queue.put(url)
    await queue.put(None)            # sentinel signals done

async def consumer(queue: asyncio.Queue, name: str) -> None:
    while True:
        url = await queue.get()
        if url is None:
            await queue.put(None)     # propagate sentinel to other consumers
            break
        result = await fetch(url)
        print(f"{name} got {url}: {len(result)} bytes")
        queue.task_done()

async def main():
    queue = asyncio.Queue(maxsize=10)
    urls = [f"https://api.example.com/{i}" for i in range(100)]
    async with asyncio.TaskGroup() as tg:
        tg.create_task(producer(queue, urls))
        for i in range(3):
            tg.create_task(consumer(queue, f"worker-{i}"))

asyncio.run(main())

Three consumers share the same queue, processing items as fast as they can. The bounded queue (maxsize=10) provides backpressure: if consumers fall behind, the producer pauses on put until space frees up. Pattern is identical to the threading version, but without GIL contention or thread overhead.

Real-World Example: Concurrent HTTP Fetches

A common production pattern: fetch many URLs concurrently with rate limiting.

import asyncio
import aiohttp

async def fetch(session: aiohttp.ClientSession, url: str) -> str:
    async with session.get(url) as response:
        return await response.text()

async def fetch_many(urls: list[str], concurrency: int = 10) -> list[str]:
    semaphore = asyncio.Semaphore(concurrency)

    async def fetch_with_limit(url: str) -> str:
        async with semaphore:
            async with aiohttp.ClientSession() as session:
                return await fetch(session, url)

    async with asyncio.TaskGroup() as tg:
        tasks = [tg.create_task(fetch_with_limit(u)) for u in urls]

    return [t.result() for t in tasks]

# Fetch 100 URLs with at most 10 concurrent requests:
results = asyncio.run(fetch_many([f"https://api.example.com/{i}" for i in range(100)]))

TaskGroup ensures all tasks complete or all are cancelled on failure; the semaphore caps concurrency. A few dozen lines, fully concurrent, error-safe, and dramatically faster than the equivalent synchronous code.

Common Mistakes

1. Forgetting await

async def main():
    result = fetch_user(1)        # WRONG: result is a coroutine, not a dict
    print(result.get("name"))      # AttributeError

async def main():
    result = await fetch_user(1)   # RIGHT
    print(result["name"])

Forgetting await produces a coroutine object instead of the awaited value. Type checkers catch many of these; RuntimeWarning: coroutine 'X' was never awaited catches the rest.

2. Mixing sync and async

Calling sync blocking code from async code freezes the loop. Always wrap blocking calls with asyncio.to_thread (for I/O) or use a process pool (for CPU work).

3. Creating coroutines without scheduling them

Naked coroutine objects (no await, no create_task) never run. The interpreter usually warns, but the warning is easy to miss.

4. Sharing state across tasks without locks

Even though async is single-threaded, race conditions exist when multiple coroutines mutate the same data across await points. Use asyncio.Lock when mutation needs to be atomic.

5. Long CPU work inside a coroutine

Even non-blocking CPU loops freeze the event loop while they run. For heavy computation, offload to a process pool or break the work into chunks separated by await asyncio.sleep(0) to yield periodically.

Frequently Asked Questions

What does await actually do in Python?

await suspends the current coroutine and returns control to the event loop. While suspended, the loop can run other ready coroutines or wait on I/O completions. When the awaited operation finishes, the loop resumes the suspended coroutine where it left off. await is NOT a blocking sleep: blocking calls would freeze the entire single-threaded loop and stop all other coroutines from making progress. The mental model: await marks the points where your code voluntarily yields the CPU back to the scheduler, which is what makes cooperative concurrency work in a single thread.

Is async/await faster than threading in Python?

For I/O-bound workloads, async/await is usually faster and uses less memory than threading because there's no thread overhead per task. A modern Python async program can manage thousands of concurrent connections in a single thread, while threading hits OS limits and context-switch costs around the hundreds. For CPU-bound work, async/await offers nothing: it's single-threaded, so it can't parallelize computation. Use async for network calls, database queries, file I/O. Use threading or multiprocessing for CPU-heavy work. The comparison is covered in detail in the next entry on choosing between concurrency models.

What is the difference between a coroutine and a task in asyncio?

A coroutine is the object created when you call an async def function: it represents pending work, but doesn't run on its own. A task is a coroutine wrapped by the event loop for scheduling: it has a name, a state, can be cancelled, and runs concurrently with other tasks. You call asyncio.create_task(coro) to convert a coroutine into a task and schedule it. Without create_task or await, a coroutine is just a paused object; the loop never runs it. The common bug: writing async_func() instead of await async_func() and wondering why nothing happens.

How do I run blocking code from within async Python?

Use asyncio.to_thread(func, *args) (Python 3.9+) to dispatch the blocking call to a thread pool. This keeps the event loop free to handle other coroutines while the blocking work runs in a worker thread. Use it for libraries that haven't been ported to async (e.g., sync database drivers, file I/O without aiofiles, third-party APIs). Common pattern: result = await asyncio.to_thread(legacy_blocking_function, arg1, arg2). For CPU-bound work, use asyncio.loop.run_in_executor() with a ProcessPoolExecutor to bypass the GIL. The honest rule: anything that doesn't accept an await keyword is suspect in an async function.

What is asyncio.TaskGroup and when should I use it?

asyncio.TaskGroup (Python 3.11+, PEP 654) is the modern structured concurrency primitive: tasks created inside the group all complete before the group exits, exceptions are collected into an ExceptionGroup, and cancellation propagates correctly. It replaces older patterns like asyncio.gather() with return_exceptions=True for most use cases. Use TaskGroup when you need to run several coroutines concurrently and want all-or-none semantics or proper exception propagation. The pattern: async with asyncio.TaskGroup() as tg: tg.create_task(work1()); tg.create_task(work2()). When the with block exits, both tasks have completed (or both have been cancelled if any raised).

The Bottom Line: One Loop, Many Coroutines, Yield to Cooperate

async/await is cooperative scheduling in a single thread. The event loop runs one coroutine at a time; await marks the points where each coroutine voluntarily yields control. async def creates a coroutine OBJECT that the loop drives; asyncio.run() is the entry point; create_task, gather, and TaskGroup are the concurrency primitives. The biggest pitfall is calling blocking code from a coroutine; the fix is asyncio.to_thread for I/O or processes for CPU work. Next up: threads vs processes vs asyncio — choosing the right concurrency model. Or browse the full Modern Python 2026 guide.