Concurrent Async Operations with Predictable Error Handling

Hello! Last time, you awaited a single HTTP request and saw how an async operation can yield while it waits for a response. This lesson addresses what happens when an incident summary needs two independent lookups: you’ll start them concurrently, collect their results, and decide what should happen if either lookup fails. Plan on about 40 minutes, including the short video segments, documentation, and a runnable example.

Starting two operations is different from awaiting them one by one

Suppose an incident summary needs its current status and its on-call owner. If each lookup waits on a different service, neither result is needed to start the other. This code nevertheless performs them sequentially:

status = await fetch_status(42)
owner = await fetch_owner(42)

The owner lookup does not begin until the status lookup finishes. Creating coroutine objects first would not change that if you then awaited the coroutines one by one. To overlap their waiting time, you need to schedule both operations as tasks.

Watch the following portions of Tech With Tim’s Asyncio in Python – Full Tutorial. The contrast between sequential awaits and scheduled tasks is the important part; you can skip the earlier introduction to coroutines, which you covered last time.

If two non-blocking lookups normally wait 0.05 and 0.30 seconds, sequential execution takes roughly their combined waiting time; concurrent execution takes roughly the longer wait, plus overhead. The event loop makes progress on another task when one yields. It does not make a blocking HTTP library or CPU-heavy calculation non-blocking.

Give concurrent work a clear lifetime

For Python 3.11 or newer, asyncio.TaskGroup is a useful default when several operations contribute to one result. Read its definition and failure rules in the official Python Coroutines and tasks documentation before running the example.

Save this as concurrent_summary.py and run it with Python 3.11 or newer. The sleeps simulate non-blocking service waits so that the timing and cancellation are easy to observe. In an application, these functions could instead await independent httpx.AsyncClient requests like the one you wrote last lesson.

import asyncio


class UpstreamUnavailable(Exception):
    pass


async def fetch_status(incident_id: int, *, fail: bool = False) -> str:
    await asyncio.sleep(0.05)
    if fail:
        raise UpstreamUnavailable(f"Status lookup failed for {incident_id}")
    return "open"


async def fetch_owner(incident_id: int) -> str:
    try:
        await asyncio.sleep(0.30)
        return "on-call"
    except asyncio.CancelledError:
        print("Owner lookup cancelled")
        raise


async def summarize(incident_id: int, *, fail_status: bool = False) -> str:
    async with asyncio.TaskGroup() as group:
        status_task = group.create_task(
            fetch_status(incident_id, fail=fail_status)
        )
        owner_task = group.create_task(fetch_owner(incident_id))

    # A normal exit means both tasks completed successfully.
    status = status_task.result()
    owner = owner_task.result()
    return f"Incident {incident_id}: {status}; owner: {owner}"


async def main() -> None:
    print(await summarize(42))

    try:
        await summarize(42, fail_status=True)
    except* UpstreamUnavailable as errors:
        for error in errors.exceptions:
            print(f"Summary unavailable: {error}")


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

On the successful call, both tasks are scheduled inside the group. Exiting the async with waits for them, so .result() is read only after successful completion. The two waits overlap; their completion order need not match the order in which you created the tasks.

On the second call, the status task fails while the owner task is still waiting. The group requests cancellation of the unfinished owner task and waits for it to finish cancelling. The owner function prints a message and re-raises CancelledError; it does not convert cancellation into a successful owner value. The group then raises an ExceptionGroup containing the upstream failure, which except* UpstreamUnavailable handles. You should see a successful summary first, followed by cancellation and failure messages. Exact timing and print interleaving can vary.

This is a fail-together policy: without a trustworthy status, the summary is not produced. Notice that the exception handler does not try to read either task’s result after the group has failed.

Choose which failures may be local

An exception group is not a promise that every task that could have failed will do so. Once a task fails, a task group cancels unfinished siblings. Two failures may appear together if they occur before cancellation takes effect, but a slower task may instead be cancelled before it reaches its own error.

Real Python’s Python 3.11 Preview: Task and Exception Groups explains how except* handles the errors that do occur. Read the indicated subsection rather than the article’s longer file-processing example.

Sometimes the owner is optional: an incident summary with a valid status and “owner unavailable” may still be useful. In that case, handle only the expected owner-lookup failure inside that task, so it does not fail the whole group:

class OwnerLookupUnavailable(Exception):
    pass


async def owner_or_none(incident_id: int) -> str | None:
    try:
        return await lookup_owner(incident_id)
    except OwnerLookupUnavailable:
        return None


async def summarize_with_optional_owner(incident_id: int) -> str:
    async with asyncio.TaskGroup() as group:
        status_task = group.create_task(fetch_status(incident_id))
        owner_task = group.create_task(owner_or_none(incident_id))

    owner = owner_task.result()
    owner_label = owner if owner is not None else "unavailable"
    return (
        f"Incident {incident_id}: {status_task.result()}; "
        f"owner: {owner_label}"
    )

Here lookup_owner() stands for your async owner-service call; the variation shows where its expected exception would be handled, not an additional runnable script. A status failure still fails the group. An expected owner failure becomes a documented fallback value. An unexpected bug in lookup_owner() still propagates rather than being silently presented as “unavailable.”

Do not catch and suppress asyncio.CancelledError as a routine fallback. Cancellation may mean the request was abandoned or another task failed; task groups rely on it to stop unfinished work. If you need cleanup on cancellation, use finally or catch CancelledError and re-raise it, as in fetch_owner().

Why not always use gather()?

asyncio.gather() also starts independent coroutines concurrently and returns successful results in argument order, regardless of which finishes first. Its failure policy differs from a task group’s, so choosing between them is not merely a style preference.

With the default return_exceptions=False, the first raised exception reaches the caller without gather() cancelling its other operations. Those operations may continue running. In a standalone script, event-loop shutdown may subsequently cancel unfinished tasks, but that is not a dependable failure policy for a long-running API server.

With return_exceptions=True, gather() waits and places exceptions alongside successful values in its result list. That can suit a deliberate collect every outcome policy, but each result must be checked before it is used as data. For the summary above, TaskGroup expresses the intended rule more directly: all required lookups succeed, or unfinished sibling work is cancelled and the failure is reported.

One operational limit applies to either approach. “Concurrent” does not mean “start an unbounded number of outbound calls.” A handful of independent lookups for one incident is different from launching requests for every incident in a large collection. Also, cancellation cannot undo an external side effect that already happened; coordinating writes requires more than choosing a task API.

Takeaways

Sequential awaits wait one after another; scheduled tasks can overlap non-blocking waits. A TaskGroup keeps related tasks within one lifetime, makes their results available after a successful exit, and cancels unfinished siblings when an unhandled task failure occurs. Handle an expected failure inside a task when a fallback is acceptable; use except* outside the group for failures that should abort the combined operation. Use gather() when its distinct result and failure behavior is what you actually need.

You can now coordinate independent async lookups. Next, you’ll place this kind of work within a clearer FastAPI structure, separating routes from application services, domain models, and infrastructure adapters.