Python asyncio: a practical mental model for safe concurrent work

When I first encountered asyncio, it felt like Python was asking me to learn a new kind of threading. It is not. asyncio is cooperative concurrency: one event loop usually runs on one OS thread, and tasks take turns whenever they reach an await that can yield control.

That distinction explains most asyncio behavior. It also explains why a program can handle many slow API calls without needing one thread per call, and why one blocking function can still freeze everything.

flowchart TD
    A[Event loop] --> B[Task A runs Python code]
    B --> C[Task A awaits I/O]
    C --> D[Event loop runs Task B]
    D --> E[Task A I/O completes]
    E --> F[Event loop resumes Task A]

The terms I need to keep straight

An async def declares a coroutine function. Calling it creates a coroutine object, but does not start its work.

async def fetch_user() -> str:
    return "Uzair"


coro = fetch_user()  # Created, not running.

async function
    call it
coroutine object
    await it or schedule it
runs on event loop

The coroutine starts only when I await it or schedule it as a Task. A Task is a coroutine registered with the event loop for independent scheduling. It is a handle for a result, an exception, and cancellation. A Task is not a thread.

task = asyncio.create_task(fetch_user())

The event loop does not interrupt Python code at random. A task must reach a yielding point, usually an await on a timer or non-blocking I/O, before another task can run.

💡 await has two meanings worth separating. The current Task cannot continue to its next line until awaited work finishes, but the event loop can still run other ready Tasks. That is very different from a blocking call such as time.sleep(1), which prevents the event-loop thread from running anything else.

await asyncio.sleep(1)  # Current Task yields.
time.sleep(1)           # Event-loop thread freezes.

Entering and leaving async code

asyncio.run(main()) is normal script code's entry point into asyncio. It creates an event loop, runs the main coroutine, cleans up unfinished tasks during shutdown, and closes the loop.

import asyncio


async def main() -> None:
    await asyncio.sleep(1)
    print("Finished")


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

normal script
    asyncio.run(main())
event loop starts
    main Task runs
event loop shuts down after main completes

Use it once at a program boundary. Inside existing async code, use await; do not start a nested asyncio.run(). The main coroutine runs as the program's root Task. During normal scheduling it behaves like other Tasks, but it defines the lifetime of the program.

Direct await versus creating a Task

Direct await is the default. It keeps dependent work in one Task and reads like ordinary code.

customer = await fetch_customer(customer_id)
orders = await fetch_orders(customer["id"])

There is no useful concurrency here because orders need the customer's ID. Use create_task() when work is independent and should overlap with other work.

customer_task = asyncio.create_task(fetch_customer(customer_id))
offers_task = asyncio.create_task(fetch_offers(customer_id))

customer = await customer_task
offers = await offers_task

Sequential
customer: [--------]
offers:             [--------]

Concurrent
customer: [--------]
offers:   [--------]

Creating a Task schedules it to run soon. It does not guarantee it begins immediately. The current Task must first yield control or finish.

💡 This is pointless concurrency:

task = asyncio.create_task(fetch_user())
user = await task

If there is no intervening independent work, write user = await fetch_user() instead.

Task lifecycle and ownership

create_task()
    pending
    runs a little Python code
    awaits I/O and pauses
    resumes when ready
    done with value, exception, or cancellation

The usual way to get a Task outcome is await task. Calling task.result() is only safe after it is done. It returns the value for a successful Task, re-raises the original exception for a failed Task, and raises CancelledError for a cancelled Task.

if task.done():
    value = task.result()

Keep a reference to every independently created Task. The reference makes ownership visible: someone can await it, cancel it during shutdown, or inspect a failure. Losing track of a task can produce the dreaded Task exception was never retrieved warning.

background_tasks: set[asyncio.Task] = set()

task = asyncio.create_task(write_audit_log())
background_tasks.add(task)
task.add_done_callback(background_tasks.discard)

This is a narrow pattern for truly independent background work. For related work, a TaskGroup is safer.

💡 asyncio.get_running_loop() answers one narrow question: "Which event loop is running this coroutine right now?" It does not create a loop, a thread, or a Task.

Coordinating concurrent work

Need every related result, fail together
    TaskGroup

Need results in input order
    gather()

Need results as they finish
    as_completed()

Need explicit done and pending control
    wait()

TaskGroup: related work succeeds or fails together

Use asyncio.TaskGroup for one coherent unit of work, such as assembling a dashboard response. It requires Python 3.11+.

async with asyncio.TaskGroup() as group:
    customer_task = group.create_task(fetch_customer())
    orders_task = group.create_task(fetch_orders())
    status_task = group.create_task(fetch_account_status())

customer = customer_task.result()
orders = orders_task.result()
status = status_task.result()

Leaving the async with block waits for every child task. If one child raises a normal exception, the TaskGroup cancels unfinished siblings, waits for their cleanup, then raises the failure, sometimes as an ExceptionGroup.

flowchart TD
    A[Dashboard request] --> B[Fetch customer]
    A --> C[Fetch orders]
    A --> D[Fetch account status]
    C --> E[Task failure]
    E --> F[Cancel unfinished sibling tasks]
    F --> G[Propagate failure to caller]

This is structured concurrency. The lifetime of child tasks is bounded by a clear scope.

gather(): collect results in input order

asyncio.gather() starts given awaitables concurrently and returns results in the same order they were supplied, even if they complete in another order.

customer, order_count = await asyncio.gather(
    fetch_customer(),
    fetch_order_count(),
)

💡 Different result types are fine. Python assigns the first returned value to customer and the second to order_count.

Its default failure behavior surprises people: if one awaitable raises, gather() raises that exception to its caller, but sibling tasks keep running. Use TaskGroup when one failure should stop related work.

return_exceptions=True collects exceptions as list values. It is useful when partial results are deliberately acceptable, but each returned item must then be inspected.

results = await asyncio.gather(
    fetch_customer(),
    fetch_recommendations(),
    return_exceptions=True,
)

as_completed() and wait()

as_completed() yields results in completion order. Use it for progress reporting, checking many endpoints, or processing the fastest successful result first.

tasks = [
    asyncio.create_task(fetch_customer()),
    asyncio.create_task(fetch_orders()),
]

for next_result in asyncio.as_completed(tasks):
    print(await next_result)

asyncio.wait() is lower-level. It returns two sets: done and pending. A timeout returns pending Tasks; it does not cancel them.

done, pending = await asyncio.wait(
    tasks,
    timeout=1.5,
    return_when=asyncio.FIRST_COMPLETED,
)

💡 done and pending are sets, not ordered lists. FIRST_COMPLETED means at least one Task has finished. More than one may be done by the time the event loop observes them.

If pending work should stop, the caller owns that decision and cleanup.

for task in pending:
    task.cancel()

await asyncio.gather(*pending, return_exceptions=True)

The * unpacks a collection into separate arguments. asyncio.gather(*pending) means the same kind of call as asyncio.gather(task_a, task_b, task_c).

Timeouts give slow work a boundary

Without a timeout, one unhealthy dependency can occupy request capacity indefinitely.

Use asyncio.wait_for() when one awaitable has its own deadline.

report = await asyncio.wait_for(fetch_report(), timeout=2)

Use asyncio.timeout() when one whole workflow shares a budget.

try:
    async with asyncio.timeout(2):
        report = await fetch_report()
        summary = await summarize(report)
except TimeoutError:
    report = "Report service exceeded its budget."

One operation deadline
    asyncio.wait_for()

Whole workflow deadline
    asyncio.timeout()

Both cancel overdue work. wait_for() waits for target cancellation cleanup, so real elapsed time may slightly exceed its numeric timeout.

Cancellation is cooperative cleanup

Calling task.cancel() requests cancellation. It does not terminate a task at an arbitrary instruction.

sequenceDiagram
    participant Caller
    participant Task
    Caller->>Task: cancel()
    Note over Task: cancellation requested
    Task->>Task: reaches next await
    Task-->>Task: CancelledError raised
    Task->>Task: finally cleanup
    Task-->>Caller: cancelled outcome

Long CPU-bound Python code with no await cannot process cancellation, and it freezes the entire event loop while it runs.

Use finally for cleanup that must happen on success, failure, or cancellation.

async def process_upload() -> None:
    connection = await open_connection()
    try:
        await upload_data(connection)
    finally:
        await connection.close()

If code catches CancelledError to perform local cleanup, it should normally re-raise it.

async def worker() -> None:
    try:
        await asyncio.sleep(10)
    except asyncio.CancelledError:
        await save_cleanup_state()
        raise

💡 await task does not cancel a Task. It waits for its final outcome. Cancellation is delivered inside the worker at that worker's next yielding point; the caller then observes the final cancelled outcome when it awaits the Task.

Avoid catching every ordinary error and quietly returning a fallback. That turns outages and programming bugs into misleading results.

try:
    payment = await charge_card()
except TimeoutError:
    return retry_later()
except CardDeclined:
    return payment_declined_result()
except asyncio.CancelledError:
    raise
except Exception:
    logger.exception("Unexpected charge failure")
    raise

CancelledError is special. Treat it as control flow: clean up and re-raise.

When shield() is justified

asyncio.shield() protects an existing Task from cancellation flowing through the caller's await. It is unusual, but can fit work such as an intentionally-owned audit write.

save_task = asyncio.create_task(write_audit_log())

try:
    await asyncio.shield(save_task)
except asyncio.CancelledError:
    # This caller remains cancelled.
    # save_task may continue under another owner.
    raise

Shield does not make the caller immune to cancellation. It also does not prevent save_task.cancel() or stop asyncio.run() shutdown from cancelling unfinished work. Keep a strong reference and make another long-lived owner responsible for its result or error.

Extra context: async I/O, blocking libraries, and CPU work

Asyncio helps only when awaited work actually yields. An async-native HTTP or database client asks the operating system to notify the event loop when I/O is ready. No worker thread waits on its behalf.

response = await client.get(url)  # Async-native client

A blocking library called from async code freezes the event loop, even if its underlying job is network I/O.

requests.get(url)  # Blocking. Do not call directly in an async endpoint.

Use asyncio.to_thread() as a bridge when an existing SDK only exposes blocking I/O.

response = await asyncio.to_thread(requests.get, url)

Blocking I/O library
    worker thread waits
    event loop stays free

Async-native I/O library
    event loop waits for OS readiness
    no worker thread required

For CPU-heavy pure-Python work, use a process pool instead. More asyncio tasks and more threads do not make Python bytecode run in parallel in a normal CPython process because of the Global Interpreter Lock, or GIL.

import asyncio
from concurrent.futures import ProcessPoolExecutor


def calculate(data: range) -> int:
    return sum(number * number for number in data)


async def main() -> None:
    loop = asyncio.get_running_loop()
    with ProcessPoolExecutor() as pool:
        result = await loop.run_in_executor(
            pool, calculate, range(1_000_000)
        )
    print(result)

Keep process-pool functions synchronous and module-level. Arguments and results cross process boundaries, so send meaningful batches rather than tiny jobs.

💡 asyncio.Future and concurrent.futures.Future are related but different families. Asyncio Futures are event-loop aware and awaitable. Executor Futures often expose a blocking .result() for synchronous callers. Application code normally awaits library-provided asyncio Futures rather than manually creating them.

yield is not await

def numbers():
    yield 1
    yield 2

yield gives the next value to a caller. await pauses a coroutine until asynchronous work is ready. A generator does not create Tasks, threads, or non-blocking I/O. Async generators can do both, yielding values while awaiting work, and consumers read them with async for.

Extra context: web-server scaling

One async server process usually has one event loop on one main thread, but it can handle many requests that are waiting on I/O.

flowchart TD
    A[Clients] --> B[Load balancer or reverse proxy]
    B --> C[Server process 1 with event loop]
    B --> D[Server process 2 with event loop]
    B --> E[Server process 3 with event loop]
    C --> F[Shared database or cache]
    D --> F
    E --> F

Multiple server workers are complete copies of the API process. They improve HTTP-serving capacity across CPU cores, but normal Python memory is not shared. Put shared sessions, caches, and coordination state in an external system such as a database, Redis, or a message broker.

Server workers
    duplicate API processes that receive requests

Process-pool workers
    calculate CPU-heavy jobs for a server process

In a common Python web stack:

Client
    Kong or another API gateway
    Uvicorn, the ASGI server
    FastAPI or Starlette application
    endpoint code using asyncio

FastAPI defines routes and application logic. Uvicorn accepts HTTP and WebSocket traffic and runs the ASGI application. Kong is an optional gateway or reverse proxy in front of services for central routing, TLS, authentication, and rate limiting. A small single service often needs only Uvicorn and the application.

My working decision guide

Need result before next line
    await coroutine directly

Independent I/O should overlap
    TaskGroup or gather()

Need each result as it finishes
    as_completed()

Need done and pending control
    wait()

One operation deadline
    wait_for()

Whole workflow deadline
    timeout()

Blocking I/O-only SDK
    to_thread()

CPU-heavy Python work
    ProcessPoolExecutor

The habit I want to build is simple: identify whether work is dependent, independent, blocking, CPU-heavy, optional, or required. Then choose the smallest asyncio primitive whose failure behavior matches that reality.

References

• Python asyncio: coroutines and tasks
• Python asyncio development and debugging
• SuperFastPython asyncio tutorial archive