Learning Hub

How to Replace asyncio.gather With TaskGroup in Python and Handle Failures With except*

Run three service calls at once in Python, watch asyncio.gather leave orphaned work after a failure, and learn how TaskGroup, ExceptionGroup and except* handle cancellation, deadlines and tests.

October 12, 2026 36 Min Read
6

What you will build, and why structured concurrency matters

Picture the product page of an online shop. To draw it, your service asks three other services at once: inventory (is the item in stock?), pricing (what does it cost today?) and reviews (what do customers think?). Asking them one after another takes as long as the three waits added together. Asking them at the same time takes as long as the slowest one. Python’s asyncio library lets you do the second thing, and starting the calls together is the easy part.

Table Of Content

The hard part is what happens when one of the calls breaks. Does the page give up at once? Do the other two calls keep running for a visitor who has already been shown an error? If two services fail, do you hear about both? Who closes the connections? These questions decide whether your service stays healthy under failure, and the two tools Python gives you answer them differently.

asyncio.gather is the older tool. asyncio.TaskGroup, added in Python 3.11, is built on an idea called structured concurrency: tasks that you start inside a block must finish before the block ends. The Python documentation describes it as “a new alternative leveraging structural concurrency; it allows for waiting for a group of related tasks with strong safety guarantees.” A widely read introduction to the idea is Nathaniel J. Smith’s essay Notes on structured concurrency, or: Go statement considered harmful, which describes the “nursery” in his Trio library. Its central rule is the one TaskGroup enforces: “Crucially, the nursery block doesn’t exit until all the tasks inside it have exited”.

In this tutorial you will run the same three-service scenario through both tools and print a timeline of what each does when a service fails. You will see gather leave work running after the caller has given up, and TaskGroup cancel it. Then you will learn to catch errors from a group with except*, see why the body of the async with is cancelled too, break cancellation on purpose in three ways (a retry helper, a blocking call and a thread), add a deadline and a concurrency limit, and combine everything in a small module with nine pytest tests. The last step shows TaskGroup.cancel(), which arrived in Python 3.15. The code uses only the standard library and pytest, and every output shown below came from running it on Python 3.13.14 on Windows 11. I also ran the demo scripts on Python 3.15.0 (the build I used is dated October 9, 2026): their output matched apart from timings and one file path in a log line.

Nine terms to know before we start

Coroutine. A function defined with async def. Calling it does not run it; it returns a coroutine object that runs only when something awaits it or schedules it.

Await. The point inside a coroutine where it pauses and lets other work run until the thing it is waiting for is ready.

Event loop. The scheduler that runs coroutines, one at a time in a single thread, switching to another coroutine whenever the current one hits an await.

Task. A coroutine wrapped so that the event loop runs it on its own, next to other tasks. You make one with asyncio.create_task() or, in this tutorial, with group.create_task().

Concurrency. Several things in progress at the same moment, taking turns while they wait. It is not parallelism: asyncio tasks take turns on one thread.

Cancellation. A request for a task to stop. asyncio delivers it by raising asyncio.CancelledError inside the task at its next await, which gives the task a chance to clean up.

Exception group. One exception object that carries several others inside it. Python’s built-in types are ExceptionGroup and BaseExceptionGroup. Their documentation says: “Both of these exception types wrap the exceptions in the sequence excs.”

Structured concurrency. The rule that a task started inside a block cannot outlive the block.

Orphan. A task that is still running after the code that started it has moved on.

Prerequisites

  • Python 3.11 or newer. asyncio.TaskGroup, asyncio.timeout() and except* all arrived in 3.11. Everything here ran on Python 3.13.14. Step 11 shows a feature that exists only in 3.15; its script also runs on older versions.
  • A first idea of async and await, as defined above. You do not need to know how the event loop works inside; the parts that matter are explained when they appear.
  • pytest 9.1.1 for the tests in Step 10. The demos need nothing beyond the standard library.
  • A terminal and an empty folder. There is no network, database or web framework in this tutorial. The “services” are small functions that wait and then answer or fail, so every run is repeatable.

Create the project

Make a folder, create a virtual environment inside it, activate it, and install pytest:

mkdir taskgroups
cd taskgroups
python -m venv .venv
.venv\Scripts\activate
python -m pip install pytest==9.1.1

On macOS and Linux the activation line is source .venv/bin/activate. Every file below goes in the taskgroups folder, and every command is run from there. On Windows, timers are coarse, so a 0.05 second wait often shows up as 0.06 seconds in the timelines. Expect differences of a few hundredths on your machine.

Step 1: Build three pretend services and call them together

Before you can compare gather and TaskGroup you need something to run. This step builds three pretend services and measures what calling them at the same time gains you.

1.1 Check your Python

Save this script and run it with python step00_versions.py. It prints your Python version and confirms that the features used in this tutorial exist.

# step00_versions.py
import asyncio
import builtins
import platform
import sys

print("Python:", sys.version.split()[0], "on", platform.system(), platform.release())
print("asyncio.TaskGroup:", hasattr(asyncio, "TaskGroup"))
print("asyncio.timeout:", hasattr(asyncio, "timeout"))
print("ExceptionGroup:", hasattr(builtins, "ExceptionGroup"))

Expected output on the machine used for this article:

Python: 3.13.14 on Windows 11
asyncio.TaskGroup: True
asyncio.timeout: True
ExceptionGroup: True

If any of the three lines says False, you are on Python 3.10 or older and need to upgrade before going on.

1.2 Write the shop module

This module is the “world” every later script lives in. Read it once, because the timelines in the rest of the tutorial come straight from it.

# shop.py
import asyncio
import time


class ReviewsDown(Exception):
    """The reviews service answered with an error."""


class PricingError(Exception):
    """The pricing service answered with an error."""


class Trace:
    """Collects timestamped events so every demo can print a timeline."""

    def __init__(self):
        self.start = time.perf_counter()
        self.events = []

    def log(self, message):
        self.events.append((time.perf_counter() - self.start, message))

    def seen(self, message):
        return any(text == message for _, text in self.events)

    def show(self):
        for seconds, text in self.events:
            print(f"  t={seconds:0.2f}s  {text}")


async def call_backend(trace, name, delay, answer, error=None):
    """Pretend to call a service: wait `delay` seconds, then return `answer` or raise `error`."""
    trace.log(f"{name}: request sent")
    try:
        await asyncio.sleep(delay)
        if error is not None:
            trace.log(f"{name}: failed")
            raise error
        trace.log(f"{name}: answered")
        return answer
    except asyncio.CancelledError:
        trace.log(f"{name}: cancelled")
        raise
    finally:
        trace.log(f"{name}: connection released")


async def inventory(trace, delay=0.20, error=None):
    return await call_backend(trace, "inventory", delay, {"in_stock": 12}, error)


async def pricing(trace, delay=0.30, error=None):
    return await call_backend(trace, "pricing", delay, {"price": 19.99}, error)


async def reviews(trace, delay=0.10, error=None):
    return await call_backend(trace, "reviews", delay, {"stars": 4.6}, error)

ReviewsDown and PricingError are two custom exception types, so later steps can tell the failures apart. Trace records each event together with the number of seconds since it was created, and show() prints the list as a timeline.

call_backend stands in for a network call. It logs “request sent”, waits delay seconds with asyncio.sleep (while it sleeps, the event loop runs other tasks, exactly as it would during real network waiting), and then either returns the answer or raises the error you passed in. The finally block logs “connection released”. That line stands for the cleanup a real client does, such as closing a socket or giving a connection back to a pool, and it is the evidence you will use all through this tutorial: it shows whether cleanup ran. The except asyncio.CancelledError branch logs “cancelled” and re-raises the exception, which is the correct way to handle cancellation; Step 6 shows what goes wrong when code does not.

The three wrappers give each service a different speed: inventory takes 0.20 seconds, pricing 0.30 and reviews 0.10.

1.3 Run them one after another, then together

The script below builds the product page both ways. asyncio.gather wraps each coroutine in a task, waits for all of them and returns the answers in the order you passed them.

# step01_sequential.py
import asyncio
import time

from shop import Trace, inventory, pricing, reviews


async def one_after_another(trace):
    return [await inventory(trace), await pricing(trace), await reviews(trace)]


async def at_the_same_time(trace):
    return await asyncio.gather(inventory(trace), pricing(trace), reviews(trace))


async def main():
    for label, build in [("one after another", one_after_another), ("at the same time", at_the_same_time)]:
        trace = Trace()
        started = time.perf_counter()
        answers = await build(trace)
        print(f"{label:>17}: {time.perf_counter() - started:0.2f}s  {answers}")


asyncio.run(main())
one after another: 0.62s  [{'in_stock': 12}, {'price': 19.99}, {'stars': 4.6}]
 at the same time: 0.31s  [{'in_stock': 12}, {'price': 19.99}, {'stars': 4.6}]

Awaiting the services one by one costs the sum of the waits (0.20 + 0.30 + 0.10 = 0.60 seconds, 0.62 measured). Running them together costs only the longest wait (0.30 seconds, 0.31 measured). That gap is the reason to bother with concurrency.

Check before moving on: your “at the same time” line should be close to 0.30 seconds and your “one after another” line close to 0.60.

Step 2: See what gather leaves behind

Now make a service fail. The Python documentation says this about gather: “If return_exceptions is False (default), the first raised exception is immediately propagated to the task that awaits on gather()”, and then adds “Other awaitables in the aws sequence won’t be cancelled and will continue to run.” Let us watch that happen.

2.1 A failure in the middle of a gather

The handler below is what a web framework would call for one visitor. Reviews fails after 0.05 seconds, while inventory and pricing are still waiting. The handler catches the error and answers with HTTP status 502. After it returns, the script counts how many tasks are still alive and then waits another 0.4 seconds so you can see what they do.

# step02_gather_orphans.py
import asyncio

from shop import ReviewsDown, Trace, inventory, pricing, reviews


async def handle_request(trace):
    """Build a product page. Return the HTTP status we would send."""
    try:
        await asyncio.gather(
            inventory(trace),
            pricing(trace),
            reviews(trace, delay=0.05, error=ReviewsDown("reviews returned 503")),
        )
    except ReviewsDown as error:
        trace.log(f"handler: caught {error!r}, sending an error page")
        return 502
    return 200


async def main():
    trace = Trace()
    status = await handle_request(trace)
    others = len(asyncio.all_tasks()) - 1
    trace.log(f"handler: returned {status}, {others} tasks are still running")
    await asyncio.sleep(0.4)
    trace.show()


asyncio.run(main())
  t=0.00s  inventory: request sent
  t=0.00s  pricing: request sent
  t=0.00s  reviews: request sent
  t=0.06s  reviews: failed
  t=0.06s  reviews: connection released
  t=0.06s  handler: caught ReviewsDown('reviews returned 503'), sending an error page
  t=0.06s  handler: returned 502, 2 tasks are still running
  t=0.20s  inventory: answered
  t=0.20s  inventory: connection released
  t=0.31s  pricing: answered
  t=0.31s  pricing: connection released

Read the timeline from the top. At 0.06 seconds reviews fails and the handler returns 502, so the visitor has their answer. The script counts two tasks still running, and the timeline shows what they do next: inventory answers at 0.20 seconds and pricing at 0.31, both for a request that finished long ago. Those are orphans. In this demo they only log a line, but in a real service they might write to a database, send an email or charge a card for a visitor who was already told the page failed. They also hold connections and CPU time that other requests could have used.

gather is doing exactly what its documentation says. The problem is that nothing in the code says that these three calls belong to this request and must stop when it ends.

2.2 The other mode: return_exceptions=True

The documentation offers an alternative: “If return_exceptions is True, exceptions are treated the same as successful results, and aggregated in the result list.” Nothing is raised, so you can see every outcome. The next script shows the cost.

# step02b_gather_values.py
import asyncio

from shop import ReviewsDown, Trace, inventory, pricing, reviews


async def main():
    trace = Trace()
    stock, price, review = await asyncio.gather(
        inventory(trace),
        pricing(trace),
        reviews(trace, delay=0.05, error=ReviewsDown("reviews returned 503")),
        return_exceptions=True,
    )
    page = {"stock": stock["in_stock"], "price": price["price"], "reviews": review}
    print("page:", page)
    try:
        print("stars:", review["stars"])
    except TypeError as error:
        print("TypeError:", error)


asyncio.run(main())
page: {'stock': 12, 'price': 19.99, 'reviews': ReviewsDown('reviews returned 503')}
TypeError: 'ReviewsDown' object is not subscriptable

The failure did not stop anything. It arrived as an ordinary value in the third slot, and the line that built the page copied that value into the reviews field without noticing, so the page dictionary now holds an error object. The next lines fail with a TypeError because they treat the exception as a dictionary (the script catches it so that you can read the message). With a template that formats values into text, nothing would be raised at all, and the error message would be printed into the web page. The error becomes data you must remember to inspect.

Step 3: Switch to a TaskGroup

A TaskGroup removes both problems. The documentation says: “Task groups combine a task creation API with a convenient and reliable way to wait for all tasks in the group to finish.” You open the group with async with, create tasks inside it, and Python will not let the block end until every task is done.

3.1 When every service answers

Save this file. build_page opens a group, creates three tasks and, after the block, reads each task’s result. The first half of main uses it when everything works; the second half makes reviews fail after 0.05 seconds.

# step03_taskgroup.py
import asyncio
import time

from shop import ReviewsDown, Trace, inventory, pricing, reviews


async def build_page(trace, reviews_delay=0.10, reviews_error=None):
    async with asyncio.TaskGroup() as group:
        stock = group.create_task(inventory(trace))
        price = group.create_task(pricing(trace))
        stars = group.create_task(reviews(trace, delay=reviews_delay, error=reviews_error))
    return {"stock": stock.result(), "price": price.result(), "reviews": stars.result()}


async def main():
    print("--- all three services answer")
    trace = Trace()
    started = time.perf_counter()
    page = await build_page(trace)
    print(f"{time.perf_counter() - started:0.2f}s  {page}")

    print("--- reviews fails after 0.05s")
    trace = Trace()
    started = time.perf_counter()
    try:
        await build_page(trace, reviews_delay=0.05, reviews_error=ReviewsDown("reviews returned 503"))
    except ExceptionGroup as group:
        print(f"{time.perf_counter() - started:0.2f}s  {type(group).__name__}: {group}")
        print("sub-exceptions:", group.exceptions)
    trace.show()
    print("tasks still running:", len(asyncio.all_tasks()) - 1)


asyncio.run(main())

Look at the success case first:

--- all three services answer
0.30s  {'stock': {'in_stock': 12}, 'price': {'price': 19.99}, 'reviews': {'stars': 4.6}}

The page took 0.30 seconds, the same as gather in Step 1. The three services ran together, and leaving the async with block waited for all of them. Notice that .result() is called after the block. A task has no result until it has finished, and the end of the block is the point where Python guarantees that.

3.2 When one service fails

Now the failure case from the same script:

--- reviews fails after 0.05s
0.06s  ExceptionGroup: unhandled errors in a TaskGroup (1 sub-exception)
sub-exceptions: (ReviewsDown('reviews returned 503'),)
  t=0.00s  inventory: request sent
  t=0.00s  pricing: request sent
  t=0.00s  reviews: request sent
  t=0.06s  reviews: failed
  t=0.06s  reviews: connection released
  t=0.06s  pricing: cancelled
  t=0.06s  pricing: connection released
  t=0.06s  inventory: cancelled
  t=0.06s  inventory: connection released
tasks still running: 0

Three things changed compared with Step 2.

  • The siblings were cancelled. The documentation says: “The first time any of the tasks belonging to the group fails with an exception other than asyncio.CancelledError, the remaining tasks in the group are cancelled.” The timeline shows pricing and inventory each logging “cancelled”, and then “connection released”, so their cleanup ran. Do not rely on the order in which cancelled siblings are listed. It is not the order in which you created the tasks (inventory was created first, yet pricing was cancelled before it), and it changed between two runs of the Step 9 script that I made.
  • The block waited for them. The group did not finish until every cancelled task had finished its cleanup. The last line, tasks still running: 0, is the opposite of Step 2’s two orphans.
  • The error arrived as an ExceptionGroup. The group did not raise ReviewsDown directly. It raised an ExceptionGroup with the message “unhandled errors in a TaskGroup (1 sub-exception)” and the real error inside, in its exceptions tuple. The documentation: those exceptions “are combined in an ExceptionGroup or BaseExceptionGroup”. That wrapping is what Step 4 is about.

Here is the comparison in one table. Every cell comes either from the Python documentation or from the timelines above.

asyncio.gather (default) asyncio.TaskGroup
When one task fails The caller gets that exception at once (0.06 s in Step 2) The other tasks are cancelled and awaited, then the block raises a group (0.06 s in Step 3)
The other tasks Keep running (they finished at 0.20 s and 0.31 s) Cancelled, and their cleanup runs before the block ends
Tasks still running afterwards 2 0
What the caller catches The original exception type An ExceptionGroup that holds it
Where the answers are A list, in argument order .result() on each task after the block

Check before moving on: in your failure run, “tasks still running” must be 0, and both siblings must show “connection released”.

Step 4: Catch errors from a group with except*

A task group delivers task failures wrapped in a group, even when only one task failed. (KeyboardInterrupt and SystemExit are the exception: the documentation says the group cancels its tasks, waits for them, and then re-raises those two as they are.) Ordinary except clauses look at the type of the group, not at the exceptions inside it. Python added a second form, except*, for this case. This step runs six small experiments, labelled A to F, in one script. Save the first part:

# step04_except_star.py
import asyncio

from shop import PricingError, ReviewsDown, Trace, inventory, pricing, reviews


async def build_page(trace, pricing_error=None, reviews_error=None):
    async with asyncio.TaskGroup() as group:
        group.create_task(inventory(trace))
        group.create_task(pricing(trace, delay=0, error=pricing_error))
        group.create_task(reviews(trace, delay=0, error=reviews_error))

build_page starts three tasks. Inventory takes its normal 0.20 seconds. Pricing and reviews are given a delay of 0 so that, if you ask them to fail, they fail in the same turn of the event loop. That makes the two-failure case in experiment C repeatable.

A. A plain except does not match

# step04_except_star.py (continued)
async def part_a():
    print("A. a plain except does not match")
    try:
        await build_page(Trace(), reviews_error=ReviewsDown("reviews returned 503"))
    except ReviewsDown:
        print("   caught by 'except ReviewsDown'")
    except Exception as error:
        print(f"   not caught by it; got {type(error).__name__}: {error}")
A. a plain except does not match
   not caught by it; got ExceptionGroup: unhandled errors in a TaskGroup (1 sub-exception)

The except ReviewsDown clause was skipped because the exception that arrived was an ExceptionGroup, which is not a ReviewsDown. Code like this passes any test that never triggers a failure, and then lets the error escape in production.

B. except* looks inside the group

# step04_except_star.py (continued)
async def part_b():
    print("B. except* matches inside the group")
    try:
        await build_page(Trace(), reviews_error=ReviewsDown("reviews returned 503"))
    except* ReviewsDown as group:
        print(f"   handler got {type(group).__name__} holding {[str(e) for e in group.exceptions]}")
    try:
        raise ReviewsDown("raised directly, no group involved")
    except* ReviewsDown as group:
        print(f"   a bare exception is wrapped for you: {group!r}")
B. except* matches inside the group
   handler got ExceptionGroup holding ['reviews returned 503']
   a bare exception is wrapped for you: ExceptionGroup('', (ReviewsDown('raised directly, no group involved'),))

The star means match against the exceptions inside the group. The variable group holds a new group containing only the matching exceptions. The language reference explains why it is always a group: when the exception raised in the try block is not a group but matches, it is wrapped “with an empty message string. This ensures that the type of the target e is consistently BaseExceptionGroup”. The second experiment in this part shows that: a bare ReviewsDown raised directly was handed to the handler inside a group. The same except* code works whether or not the error travelled inside a group.

C. Two failures, two handlers

# step04_except_star.py (continued)
async def part_c():
    print("C. two failures, two handlers run")
    try:
        await build_page(
            Trace(),
            pricing_error=PricingError("bad currency"),
            reviews_error=ReviewsDown("reviews returned 503"),
        )
    except* PricingError as group:
        print(f"   pricing handler: {[str(e) for e in group.exceptions]}")
    except* ReviewsDown as group:
        print(f"   reviews handler: {[str(e) for e in group.exceptions]}")
C. two failures, two handlers run
   pricing handler: ['bad currency']
   reviews handler: ['reviews returned 503']

With a normal try statement at most one except clause runs. Here both ran, each receiving the part of the group that matches it. This is how you hear about every failure that happened, not just the first.

D. What no handler matches keeps travelling

# step04_except_star.py (continued)
async def part_d():
    print("D. whatever no handler matches keeps travelling")
    try:
        try:
            await build_page(
                Trace(),
                pricing_error=PricingError("bad currency"),
                reviews_error=ReviewsDown("reviews returned 503"),
            )
        except* ReviewsDown:
            print("   inner handler took the reviews error")
    except ExceptionGroup as rest:
        print(f"   outer code received {type(rest).__name__} holding {[repr(e) for e in rest.exceptions]}")
D. whatever no handler matches keeps travelling
   inner handler took the reviews error
   outer code received ExceptionGroup holding ["PricingError('bad currency')"]

Only the reviews error was handled. The pricing error was not lost: it left the try statement in a new group for the outer code to deal with. The documentation says: “After all except* clauses execute, the group of unhandled exceptions is merged with any exceptions that were raised or re-raised from within except* clauses.”

E. Raising a new exception inside except*

# step04_except_star.py (continued)
async def part_e():
    print("E. raising a new exception inside except*")
    for label, pricing_error in [("nothing left over", None), ("a PricingError left over", PricingError("bad currency"))]:
        try:
            try:
                await build_page(
                    Trace(),
                    pricing_error=pricing_error,
                    reviews_error=ReviewsDown("reviews returned 503"),
                )
            except* ReviewsDown as group:
                raise RuntimeError("page unavailable") from group
        except BaseException as caught:
            if isinstance(caught, BaseExceptionGroup):
                print(f"   {label}: the caller received {type(caught).__name__} holding {[repr(e) for e in caught.exceptions]}")
            else:
                print(f"   {label}: the caller received {caught!r}, not wrapped")
E. raising a new exception inside except*
   nothing left over: the caller received RuntimeError('page unavailable'), not wrapped
   a PricingError left over: the caller received ExceptionGroup holding ["RuntimeError('page unavailable')", "ExceptionGroup('unhandled errors in a TaskGroup', [PricingError('bad currency')])"]

The result depends on what else was in the group. When nothing was left over, the new RuntimeError arrived on its own, which matches PEP 654: “Raising a new instance of a naked exception does not cause this exception to be wrapped by an exception group.” When a PricingError was left over, the new exception and the leftover group travelled together in one new group, so the caller received an ExceptionGroup. A caller cannot predict which one it will get. In Step 9 you will therefore convert group errors into your own exception with a plain except ExceptionGroup at the edge of your code, not by raising from inside except*.

F. Two syntax rules

# step04_except_star.py (continued)
RETURN_IN_EXCEPT_STAR = """
def f():
    try:
        pass
    except* ValueError:
        return 1
"""

MIXED_HANDLERS = """
try:
    pass
except ValueError:
    pass
except* TypeError:
    pass
"""


def part_f():
    print("F. except* has two syntax rules")
    for label, source in [
        ("return inside except*", RETURN_IN_EXCEPT_STAR),
        ("except mixed with except*", MIXED_HANDLERS),
    ]:
        try:
            compile(source, "<demo>", "exec")
        except SyntaxError as error:
            print(f"   {label}: SyntaxError: {error.msg}")
# step04_except_star.py (continued)
async def main():
    await part_a()
    await part_b()
    await part_c()
    await part_d()
    await part_e()
    part_f()


asyncio.run(main())
F. except* has two syntax rules
   return inside except*: SyntaxError: 'break', 'continue' and 'return' cannot appear in an except* block
   except mixed with except*: SyntaxError: cannot have both 'except' and 'except*' on the same 'try'

A single try statement uses either except or except*, never both, and break, continue and return are not allowed inside an except* block. The script uses compile() on two small snippets so that you can see the exact error messages without breaking your own file. In practice, set a variable inside the block and act on it after the try statement.

Reading an uncaught group

If nothing catches the group, Python prints it as a tree. This tiny script fails on purpose:

# step04a_uncaught.py
import asyncio

from shop import ReviewsDown, Trace, inventory, reviews


async def main():
    trace = Trace()
    async with asyncio.TaskGroup() as group:
        group.create_task(inventory(trace))
        group.create_task(reviews(trace, delay=0.05, error=ReviewsDown("reviews returned 503")))


asyncio.run(main())
  + Exception Group Traceback (most recent call last):
  |   File "step04a_uncaught.py", line 14, in <module>
  |     asyncio.run(main())
  |     ~~~~~~~~~~~^^^^^^^^
  |   File "<python>\Lib\asyncio\runners.py", line 196, in run
  |     return runner.run(main)
  |            ~~~~~~~~~~^^^^^^
  |   File "<python>\Lib\asyncio\runners.py", line 119, in run
  |     return self._loop.run_until_complete(task)
  |            ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~^^^^^^
  |   File "<python>\Lib\asyncio\base_events.py", line 725, in run_until_complete
  |     return future.result()
  |            ~~~~~~~~~~~~~^^
  |   File "step04a_uncaught.py", line 9, in main
  |     async with asyncio.TaskGroup() as group:
  |                ~~~~~~~~~~~~~~~~~^^
  |   File "<python>\Lib\asyncio\taskgroups.py", line 71, in __aexit__
  |     return await self._aexit(et, exc)
  |            ^^^^^^^^^^^^^^^^^^^^^^^^^^
  |   File "<python>\Lib\asyncio\taskgroups.py", line 173, in _aexit
  |     raise BaseExceptionGroup(
  |     ...<2 lines>...
  |     ) from None
  | ExceptionGroup: unhandled errors in a TaskGroup (1 sub-exception)
  +-+---------------- 1 ----------------
    | Traceback (most recent call last):
    |   File "shop.py", line 58, in reviews
    |     return await call_backend(trace, "reviews", delay, {"stars": 4.6}, error)
    |            ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
    |   File "shop.py", line 39, in call_backend
    |     raise error
    | shop.ReviewsDown: reviews returned 503
    +------------------------------------

In the output I removed the folder prefix from file paths. The top part, above the line of dashes and plus signs, is the traceback of the group itself: it shows where the group was raised, at the async with asyncio.TaskGroup() line, and ends with the group’s message. Each failed task then gets its own numbered box underneath, with its own traceback. Here box 1 shows that ReviewsDown was raised by raise error inside call_backend. When a group holds three errors, you get three boxes. Read the boxes first, because that is where your real error is.

Step 5: Remember the body is cancelled too

The code inside the async with block, outside any task, is also part of the group. The documentation says that when a task fails and the body is still running, “the task directly containing the async with statement is also cancelled. The resulting asyncio.CancelledError will interrupt an await, but it will not bubble out of the containing async with statement.” This script gives the body ten seconds of work and watches what happens.

# step05_body_cancelled.py
import asyncio

from shop import ReviewsDown, Trace, reviews


async def main():
    trace = Trace()
    try:
        async with asyncio.TaskGroup() as group:
            group.create_task(reviews(trace, delay=0.05, error=ReviewsDown("reviews returned 503")))
            try:
                trace.log("body: starting 10 seconds of other work")
                await asyncio.sleep(10)
                trace.log("body: finished its other work")
            except asyncio.CancelledError:
                trace.log("body: interrupted by cancellation")
                raise
            finally:
                trace.log("body: finally block ran")
    except* ReviewsDown:
        trace.log("outside: caught ReviewsDown, no CancelledError escaped")
    trace.show()


asyncio.run(main())
  t=0.00s  body: starting 10 seconds of other work
  t=0.00s  reviews: request sent
  t=0.07s  reviews: failed
  t=0.07s  reviews: connection released
  t=0.07s  body: interrupted by cancellation
  t=0.07s  body: finally block ran
  t=0.07s  outside: caught ReviewsDown, no CancelledError escaped

The body logged that it was starting ten seconds of work, but reviews failed at 0.06 seconds and the body’s await asyncio.sleep(10) was interrupted at once. Its finally block ran, and the CancelledError did not escape: the code after the async with saw only ReviewsDown. This is the behaviour you want, because a request that has failed should not sit through ten seconds of leftover work. It also means that the body must be written like a task: use try/finally for cleanup, and do not swallow CancelledError.

Step 6: Write code that cooperates with cancellation

Cancellation is a request, not a kill. The documentation for Task.cancel() says the coroutine “has a chance to clean up or even deny the request by suppressing the exception”, and that “suppressing cancellation completely is not common and is actively discouraged.” A task group depends on tasks cooperating, and the docs warn that the components that enable structured concurrency, such as TaskGroup and asyncio.timeout(), “might misbehave if a coroutine swallows asyncio.CancelledError”. This step breaks cooperation in three ways so that you can recognise each one.

6.1 A retry helper that swallows cancellation

Retry helpers are everywhere. This one has a classic bug: it catches BaseException, and asyncio.CancelledError is a BaseException. The documentation notes that CancelledError “directly subclasses BaseException so most code will not need to be aware of it”, and that is exactly why except Exception is safe and except BaseException is not.

# step06_swallow.py
import asyncio
import time

from shop import ReviewsDown, Trace, inventory, reviews


async def retry_buggy(make_call, attempts=3):
    """Retry a flaky call. BUG: BaseException also catches asyncio.CancelledError."""
    for attempt in range(1, attempts + 1):
        try:
            return await make_call()
        except BaseException:
            if attempt == attempts:
                raise


async def retry_fixed(make_call, attempts=3):
    """Same helper, but only ordinary errors are retried. CancelledError passes straight through."""
    for attempt in range(1, attempts + 1):
        try:
            return await make_call()
        except Exception:
            if attempt == attempts:
                raise


async def build_page(trace, retry):
    async with asyncio.TaskGroup() as group:
        group.create_task(retry(lambda: inventory(trace)))
        group.create_task(reviews(trace, delay=0.05, error=ReviewsDown("reviews returned 503")))


async def main():
    for label, retry in [("buggy retry helper", retry_buggy), ("fixed retry helper", retry_fixed)]:
        trace = Trace()
        started = time.perf_counter()
        try:
            await build_page(trace, retry)
        except* ReviewsDown:
            pass
        print(f"--- {label}: the group finished after {time.perf_counter() - started:0.2f}s")
        trace.show()


asyncio.run(main())
--- buggy retry helper: the group finished after 0.27s
  t=0.00s  inventory: request sent
  t=0.00s  reviews: request sent
  t=0.06s  reviews: failed
  t=0.06s  reviews: connection released
  t=0.06s  inventory: cancelled
  t=0.06s  inventory: connection released
  t=0.06s  inventory: request sent
  t=0.27s  inventory: answered
  t=0.27s  inventory: connection released
--- fixed retry helper: the group finished after 0.06s
  t=0.00s  inventory: request sent
  t=0.00s  reviews: request sent
  t=0.06s  reviews: failed
  t=0.06s  reviews: connection released
  t=0.06s  inventory: cancelled
  t=0.06s  inventory: connection released

With the buggy helper, inventory was cancelled at 0.06 seconds, as the timeline shows. The helper treated the cancellation as a failed attempt and started attempt two. That attempt ran to completion and logged “answered” at 0.27 seconds, so the group stayed open for 0.27 seconds instead of 0.06, doing work for a request that had already failed. With more attempts or longer waits, the delay grows to match. With except Exception the group finished at 0.06 seconds with inventory cancelled and released.

A real retry helper also needs a delay between attempts, which is covered in the exponential backoff and jitter tutorial. Whatever you write, let CancelledError pass through.

6.2 A blocking call freezes the whole loop

The event loop runs everything in one thread, so a call that blocks, such as time.sleep, a slow library call or a big file read, stops every task until it returns. No task can be cancelled while the loop is stuck. The script below runs a 0.5 second blocking call next to the failing reviews service, first directly and then through asyncio.to_thread, which runs a function in a separate thread. It turns on asyncio’s debug mode with debug=True.

# step06b_blocking.py
import asyncio
import logging
import threading
import time

from shop import ReviewsDown, Trace, reviews

logging.basicConfig(level=logging.WARNING, format="  asyncio log: %(message)s")


def slow_report(trace):
    """A blocking call, like a legacy library or a big file read."""
    time.sleep(0.5)
    trace.log(f"report: blocking work finished in thread {threading.current_thread().name}")


async def report_blocking(trace):
    slow_report(trace)


async def report_in_thread(trace):
    await asyncio.to_thread(slow_report, trace)


async def build_page(trace, report):
    async with asyncio.TaskGroup() as group:
        group.create_task(reviews(trace, delay=0.05, error=ReviewsDown("reviews returned 503")))
        group.create_task(report(trace))


async def main():
    for label, report in [
        ("blocking call inside the task", report_blocking),
        ("the same call in a thread", report_in_thread),
    ]:
        trace = Trace()
        started = time.perf_counter()
        try:
            await build_page(trace, report)
        except* ReviewsDown:
            trace.log("outside: caught ReviewsDown")
        print(f"--- {label}: the group finished after {time.perf_counter() - started:0.2f}s")
        await asyncio.sleep(0.6)
        trace.show()


asyncio.run(main(), debug=True)
  asyncio log: Executing <Task finished name='Task-3' coro=<report_blocking() done, defined at step06b_blocking.py:18> result=None created at <python>\Lib\asyncio\taskgroups.py:196> took 0.501 seconds
--- blocking call inside the task: the group finished after 0.50s
  t=0.00s  reviews: request sent
  t=0.50s  report: blocking work finished in thread MainThread
  t=0.50s  reviews: failed
  t=0.50s  reviews: connection released
  t=0.50s  outside: caught ReviewsDown
--- the same call in a thread: the group finished after 0.06s
  t=0.00s  reviews: request sent
  t=0.06s  reviews: failed
  t=0.06s  reviews: connection released
  t=0.06s  outside: caught ReviewsDown
  t=0.51s  report: blocking work finished in thread asyncio_0

In the first run, reviews should have failed at 0.05 seconds, but the blocking call held the loop for 0.5 seconds, so the failure was not even noticed until 0.50 seconds. In the second run the blocking function ran in a thread, and the group finished at 0.06 seconds.

The line at the top is asyncio’s debug mode at work. The documentation lists several ways to turn it on (the PYTHONASYNCIODEBUG environment variable, Python Development Mode, debug=True and loop.set_debug()) and says: “Callbacks taking longer than 100 milliseconds are logged.” The warning names the task, shows the function (report_blocking()) and says it took 0.501 seconds. I removed the folder prefix from the file paths. When a service feels slow for no clear reason, turn on debug mode in a test environment and look for these lines.

6.3 A thread keeps running after cancellation

Look at the last line of the output above. In the second run the group finished at 0.06 seconds, but the thread logged “blocking work finished” at 0.51 seconds, long after the group was gone. Cancelling a task that is waiting for a thread stops the waiting. It cannot stop the thread. That is harmless for a read-only report and a real problem for a thread that writes data or holds a lock, so give such functions their own way to stop, such as a flag they check, and do not assume that TaskGroup will stop them.

Check before moving on: your buggy helper should keep the group open for about 0.27 seconds and the fixed one for about 0.06, and your blocking run should take about 0.5 seconds against about 0.06 for the threaded one.

Step 7: Add a deadline with asyncio.timeout

A page that waits forever for a slow service is as bad as one that fails. asyncio.timeout() is a context manager that cancels the code inside it when time runs out and turns the cancellation into a TimeoutError. Where you put it changes what you catch, so the script below tries two placements: around the whole group, and inside one child.

# step07_timeouts.py
import asyncio
import time

from shop import Trace, inventory, pricing, reviews


async def limited(awaitable, seconds):
    async with asyncio.timeout(seconds):
        return await awaitable


async def build_page(trace, pricing_limit=None):
    async with asyncio.TaskGroup() as group:
        group.create_task(inventory(trace))
        if pricing_limit is None:
            group.create_task(pricing(trace))
        else:
            group.create_task(limited(pricing(trace), pricing_limit))
        group.create_task(reviews(trace))


async def main():
    print("--- one deadline around the whole group: 0.25s, but pricing needs 0.30s")
    trace = Trace()
    started = time.perf_counter()
    try:
        async with asyncio.timeout(0.25):
            await build_page(trace)
    except TimeoutError:
        print(f"a plain 'except TimeoutError' caught it after {time.perf_counter() - started:0.2f}s")
    trace.show()

    print("--- a deadline inside one child: pricing gets 0.15s")
    trace = Trace()
    started = time.perf_counter()
    try:
        await build_page(trace, pricing_limit=0.15)
    except* TimeoutError as group:
        print(f"except* caught {group.exceptions!r} after {time.perf_counter() - started:0.2f}s")
    trace.show()


asyncio.run(main())
--- one deadline around the whole group: 0.25s, but pricing needs 0.30s
a plain 'except TimeoutError' caught it after 0.26s
  t=0.00s  inventory: request sent
  t=0.00s  pricing: request sent
  t=0.00s  reviews: request sent
  t=0.10s  reviews: answered
  t=0.10s  reviews: connection released
  t=0.22s  inventory: answered
  t=0.22s  inventory: connection released
  t=0.26s  pricing: cancelled
  t=0.26s  pricing: connection released
--- a deadline inside one child: pricing gets 0.15s
except* caught (TimeoutError(),) after 0.16s
  t=0.00s  inventory: request sent
  t=0.00s  pricing: request sent
  t=0.00s  reviews: request sent
  t=0.11s  reviews: answered
  t=0.11s  reviews: connection released
  t=0.16s  pricing: cancelled
  t=0.16s  pricing: connection released
  t=0.16s  inventory: cancelled
  t=0.16s  inventory: connection released

In the first run, pricing needed 0.30 seconds and the deadline was 0.25. At the deadline the only task still running was pricing; it was cancelled, its connection was released, and a plain except TimeoutError caught the result. The documentation explains why that works: “The asyncio.timeout() context manager is what transforms the asyncio.CancelledError into a TimeoutError, which means the TimeoutError can only be caught outside of the context manager.” Here the context manager is outside the group, so you catch the error outside both.

In the second run the deadline sat inside one child. Pricing raised TimeoutError after 0.15 seconds like any other failing task, so the group cancelled inventory and delivered the error inside an ExceptionGroup. A plain except TimeoutError would have missed it, so the script uses except*.

Where the deadline sits What happens when it expires What you write
Around the whole TaskGroup Every running task in the group is cancelled, and its cleanup runs except TimeoutError
Inside one child task That child fails; the group cancels its siblings except* TimeoutError

Step 8: Limit how many tasks run at once

So far you started three tasks. Real code often loops over a list, such as 200 product pages to refresh. A task group starts every task you create right away. If the service behind them has a limit, such as a database with a connection cap or an API with a rate limit, flooding it makes it refuse requests.

The Semaphore is the standard tool. The documentation says: “A semaphore manages an internal counter which is decremented by each acquire() call and incremented by each release() call.” In other words it hands out a fixed number of permits; async with gate: takes one, waits if none are left, and gives it back at the end, even when the task fails or is cancelled. Save this helper, which turns the idea into a reusable function:

# bounded.py
import asyncio


async def bounded_map(function, items, *, limit):
    """Call `await function(item)` for every item, at most `limit` at a time.

    Results keep the order of `items`. The first failure cancels everything else.
    """
    gate = asyncio.Semaphore(limit)

    async def guarded(item):
        async with gate:
            return await function(item)

    async with asyncio.TaskGroup() as group:
        tasks = [group.create_task(guarded(item)) for item in items]
    return [task.result() for task in tasks]

All the tasks are created, but only limit of them get past the gate at any moment. Results keep the order of items, and one failure still cancels the rest. The script below compares it with the unbounded version against a pretend server that holds 10 requests at once and refuses any beyond that:

# step08_bounded.py
import asyncio
import time

from bounded import bounded_map


class ServerBusy(Exception):
    """The pretend server refused a request because it was already full."""


class FakeServer:
    """Counts requests in flight and refuses any that arrive while `capacity` are already running."""

    def __init__(self, capacity, latency):
        self.capacity = capacity
        self.latency = latency
        self.in_flight = 0
        self.peak = 0

    async def get(self, item):
        self.in_flight += 1
        self.peak = max(self.peak, self.in_flight)
        try:
            await asyncio.sleep(0)  # the handshake: other requests get to arrive meanwhile
            if self.in_flight > self.capacity:
                raise ServerBusy(f"item {item}: server full")
            await asyncio.sleep(self.latency)
            return item * 2
        finally:
            self.in_flight -= 1


async def fetch_all_unbounded(server, items):
    async with asyncio.TaskGroup() as group:
        tasks = [group.create_task(server.get(item)) for item in items]
    return [task.result() for task in tasks]


async def main():
    items = list(range(200))

    server = FakeServer(capacity=10, latency=0.05)
    started = time.perf_counter()
    try:
        await fetch_all_unbounded(server, items)
    except* ServerBusy as group:
        print(
            f"unbounded: {len(group.exceptions)} requests refused, peak in flight {server.peak}, "
            f"{time.perf_counter() - started:0.2f}s, no answers at all"
        )

    server = FakeServer(capacity=10, latency=0.05)
    started = time.perf_counter()
    answers = await bounded_map(server.get, items, limit=10)
    print(
        f"bounded:   {len(answers)} answers, peak in flight {server.peak}, "
        f"{time.perf_counter() - started:0.2f}s, first five {answers[:5]}"
    )


asyncio.run(main())
unbounded: 190 requests refused, peak in flight 200, 0.00s, no answers at all
bounded:   200 answers, peak in flight 10, 1.27s, first five [0, 2, 4, 6, 8]

Without a limit, all 200 requests arrived together, the server refused 190 of them, the peak was 200 requests in flight, and the first refusal cancelled the whole batch, so there were no answers at all. With bounded_map the peak never passed 10 and all 200 requests were answered, in order. The ideal time is 20 rounds of 0.05 seconds, which is 1.0 second; the measured 1.27 seconds includes the small timer overheads seen in earlier steps. The semaphore is the simplest form of the bulkhead pattern. If you need a service to refuse extra work instead of queueing it, read the backpressure and load shedding tutorial.

Step 9: Put it together

Now combine the pieces into the function that a web handler would call. It follows four rules that come straight from the earlier steps:

  1. Inventory and pricing are required: if either fails, the page fails and everything else stops.
  2. Reviews are optional: if they fail, the page is built without them and marked as degraded.
  3. The whole page has a deadline.
  4. Callers get one kind of error, PageUnavailable, rather than an ExceptionGroup.
# product_page.py
import asyncio

from shop import inventory, pricing, reviews


class PageUnavailable(Exception):
    """The product page could not be built. `causes` holds the underlying errors, flattened."""

    def __init__(self, message, causes=()):
        super().__init__(message)
        self.causes = list(causes)


def leaves(error):
    """Yield every ordinary exception inside an ExceptionGroup, however deeply nested."""
    if isinstance(error, BaseExceptionGroup):
        for inner in error.exceptions:
            yield from leaves(inner)
    else:
        yield error


async def optional(awaitable, *, default, trace, name):
    """Run a nice-to-have call. Ordinary failures become `default`; cancellation still passes through."""
    try:
        return await awaitable
    except Exception as error:
        trace.log(f"{name}: optional call failed ({error!r}), using the default")
        return default


async def load_product_page(
    trace,
    *,
    deadline=0.5,
    fetch_inventory=inventory,
    fetch_pricing=pricing,
    fetch_reviews=reviews,
):
    """Build the page from three services. Inventory and pricing are required, reviews are optional."""
    try:
        async with asyncio.timeout(deadline):
            async with asyncio.TaskGroup() as group:
                stock = group.create_task(fetch_inventory(trace))
                price = group.create_task(fetch_pricing(trace))
                stars = group.create_task(optional(fetch_reviews(trace), default=None, trace=trace, name="reviews"))
    except TimeoutError:
        raise PageUnavailable(f"no answer within {deadline}s") from None
    except ExceptionGroup as group:
        raise PageUnavailable("a required service failed", leaves(group)) from group
    return {
        "stock": stock.result(),
        "price": price.result(),
        "reviews": stars.result(),
        "degraded": stars.result() is None,
    }

optional catches Exception, not BaseException, for the reason in Step 6: a cancelled reviews call must still be cancelled. A failure becomes a default value, so the group never sees it, and the siblings keep running. leaves walks a group, including groups nested inside groups, and yields the ordinary exceptions, so PageUnavailable.causes is a flat list that you can log.

In load_product_page the deadline sits outside the group, which is why except TimeoutError works (Step 7). The next clause, except ExceptionGroup, is a plain except on purpose. We want to convert whatever arrived into our own exception, and Step 4E showed that raising from inside except* gives callers a different type depending on what else was in the group. The functions for the three services are parameters, which lets tests and demos replace them with failing versions without any patching. Save the demo script and run it:

# step09_page_demo.py
import asyncio
import functools
import time

from product_page import PageUnavailable, load_product_page
from shop import PricingError, ReviewsDown, Trace, inventory, pricing, reviews


async def show(label, **services):
    trace = Trace()
    started = time.perf_counter()
    try:
        page = await load_product_page(trace, **services)
        outcome = f"page {page}"
    except PageUnavailable as error:
        causes = [repr(cause) for cause in error.causes]
        outcome = f"PageUnavailable({str(error)!r}) causes={causes}"
    print(f"{label}: {time.perf_counter() - started:0.2f}s  {outcome}")
    return trace


async def main():
    await show("1 all services answer")
    await show(
        "2 reviews down (optional)",
        fetch_reviews=functools.partial(reviews, delay=0.05, error=ReviewsDown("reviews returned 503")),
    )
    trace = await show(
        "3 pricing down (required)",
        fetch_pricing=functools.partial(pricing, delay=0.05, error=PricingError("bad currency")),
    )
    trace.show()
    await show(
        "4 pricing and inventory down",
        fetch_inventory=functools.partial(inventory, delay=0, error=RuntimeError("inventory timed out")),
        fetch_pricing=functools.partial(pricing, delay=0, error=PricingError("bad currency")),
    )
    await show("5 pricing too slow for a 0.25s deadline", deadline=0.25)


asyncio.run(main())
1 all services answer: 0.31s  page {'stock': {'in_stock': 12}, 'price': {'price': 19.99}, 'reviews': {'stars': 4.6}, 'degraded': False}
2 reviews down (optional): 0.30s  page {'stock': {'in_stock': 12}, 'price': {'price': 19.99}, 'reviews': None, 'degraded': True}
3 pricing down (required): 0.06s  PageUnavailable('a required service failed') causes=["PricingError('bad currency')"]
  t=0.00s  inventory: request sent
  t=0.00s  pricing: request sent
  t=0.00s  reviews: request sent
  t=0.06s  pricing: failed
  t=0.06s  pricing: connection released
  t=0.06s  reviews: cancelled
  t=0.06s  reviews: connection released
  t=0.06s  inventory: cancelled
  t=0.06s  inventory: connection released
4 pricing and inventory down: 0.00s  PageUnavailable('a required service failed') causes=["RuntimeError('inventory timed out')", "PricingError('bad currency')"]
5 pricing too slow for a 0.25s deadline: 0.26s  PageUnavailable('no answer within 0.25s') causes=[]

Scenario 1 shows the happy path in 0.31 seconds. In scenario 2 the reviews service failed, and the page was still built in 0.30 seconds, with reviews set to None and degraded set to True. In scenario 3 pricing failed after 0.05 seconds, and the timeline shows inventory and reviews cancelled and released: one required failure ended the request at once, with no orphans. In scenario 4 two required services failed together and causes lists both. In scenario 5 pricing needed 0.30 seconds against a 0.25 second deadline, so the caller got PageUnavailable with the message “no answer within 0.25s”.

Step 10: Test it with pytest

Behaviour like “siblings are cancelled and cleaned up” is easy to break during a refactor, so pin it down with tests. You do not need a plugin: each test calls asyncio.run() on the coroutine it wants to check. The timeline lines recorded by Trace make the cleanup assertions simple. Save the first half of the tests:

# test_product_page.py
import asyncio
import functools
import time

import pytest

from bounded import bounded_map
from product_page import PageUnavailable, leaves, load_product_page, optional
from shop import PricingError, ReviewsDown, Trace, inventory, pricing, reviews


def test_all_services_answer_at_the_same_time():
    trace = Trace()
    started = time.perf_counter()
    page = asyncio.run(load_product_page(trace))
    elapsed = time.perf_counter() - started
    assert page == {
        "stock": {"in_stock": 12},
        "price": {"price": 19.99},
        "reviews": {"stars": 4.6},
        "degraded": False,
    }
    assert elapsed < 0.55  # one after another would take 0.60s


def test_a_failed_optional_call_gives_a_degraded_page():
    trace = Trace()
    broken = functools.partial(reviews, delay=0.05, error=ReviewsDown("reviews returned 503"))
    page = asyncio.run(load_product_page(trace, fetch_reviews=broken))
    assert page["reviews"] is None
    assert page["degraded"] is True
    assert trace.seen("reviews: failed")


def test_a_failed_required_call_cancels_the_others_and_cleans_up():
    trace = Trace()
    broken = functools.partial(pricing, delay=0.05, error=PricingError("bad currency"))
    started = time.perf_counter()
    with pytest.raises(PageUnavailable) as caught:
        asyncio.run(load_product_page(trace, fetch_pricing=broken))
    assert time.perf_counter() - started < 0.2
    assert [type(cause) for cause in caught.value.causes] == [PricingError]
    assert trace.seen("inventory: cancelled")
    assert trace.seen("inventory: connection released")
    assert not trace.seen("inventory: answered")


def test_every_required_failure_is_reported():
    trace = Trace()
    bad_inventory = functools.partial(inventory, delay=0, error=RuntimeError("inventory timed out"))
    bad_pricing = functools.partial(pricing, delay=0, error=PricingError("bad currency"))
    with pytest.raises(PageUnavailable) as caught:
        asyncio.run(load_product_page(trace, fetch_inventory=bad_inventory, fetch_pricing=bad_pricing))
    assert {type(cause) for cause in caught.value.causes} == {RuntimeError, PricingError}


def test_the_deadline_cancels_everything_and_leaves_no_tasks_behind():
    async def scenario():
        trace = Trace()
        with pytest.raises(PageUnavailable, match="no answer within"):
            await load_product_page(trace, deadline=0.15)
        return trace, len(asyncio.all_tasks())

    trace, running = asyncio.run(scenario())
    assert running == 1  # only the scenario itself
    assert trace.seen("pricing: cancelled")

The third test is the heart of the file. It makes pricing fail at 0.05 seconds, and then checks that the error reached the caller as PageUnavailable with the right cause, that the whole call took under 0.2 seconds, and that inventory was cancelled and released and never answered. The fifth test sets a 0.15 second deadline and checks that, afterwards, the only task alive is the test’s own, so nothing was left behind. Now the second half:

# test_product_page.py (continued)
def test_optional_lets_cancellation_through():
    async def scenario():
        task = asyncio.create_task(optional(asyncio.sleep(5), default=None, trace=Trace(), name="slow"))
        await asyncio.sleep(0)
        task.cancel()
        with pytest.raises(asyncio.CancelledError):
            await task

    asyncio.run(scenario())


def test_leaves_flattens_nested_groups():
    inner = ExceptionGroup("inner", [ValueError("a"), TypeError("b")])
    outer = ExceptionGroup("outer", [inner, KeyError("c")])
    assert sorted(type(error).__name__ for error in leaves(outer)) == ["KeyError", "TypeError", "ValueError"]


def test_bounded_map_keeps_order_and_respects_the_limit():
    state = {"running": 0, "peak": 0}

    async def work(item):
        state["running"] += 1
        state["peak"] = max(state["peak"], state["running"])
        await asyncio.sleep(0.01)
        state["running"] -= 1
        return item * 2

    answers = asyncio.run(bounded_map(work, range(30), limit=4))
    assert answers == [item * 2 for item in range(30)]
    assert state["peak"] == 4


def test_bounded_map_stops_starting_work_after_a_failure():
    started = []

    async def work(item):
        started.append(item)
        await asyncio.sleep(0.02)
        if item == 3:
            raise ValueError("item 3 failed")
        return item

    with pytest.raises(ExceptionGroup):
        asyncio.run(bounded_map(work, range(40), limit=2))
    assert len(started) < 10

The sixth test guards the rule from Step 6: it cancels a task that is inside optional and demands that the cancellation comes out. The seventh test checks leaves against a group nested inside a group. The last two cover bounded_map: the peak of concurrent work equals the limit, answers keep their order, and after a failure most of the 40 items never start. Run the suite:

python -m pytest -q
......... [100%]
9 passed in 1.10s

All nine pass. The time thresholds are generous on purpose (for example, a successful page must take under 0.55 seconds, where the sequential version would need about 0.60), because tests that depend on a tight clock fail at random on busy machines.

Step 11: Stop a group early on Python 3.15

Sometimes a group should end early without any error. Imagine asking three replicas of a service the same question and keeping the first answer. Before Python 3.15 the usual way was to raise a private exception from one task and catch it outside with except*. Python 3.15 added TaskGroup.cancel(). The documentation calls it “a non-exceptional, early exit of the task group’s lifetime” that is “useful once the group’s goal has been met or its services no longer needed”, and promises that “The task group context manager will exit without asyncio.CancelledError being raised.” The script below uses cancel() when the group has it and falls back to the private exception when it does not, so it runs on both:

# step11_first_answer.py
import asyncio
import sys
import time

from shop import Trace, call_backend


class GotAnswer(Exception):
    """Used on Python 3.14 and older to leave a TaskGroup early."""


async def first_answer(trace):
    """Ask three replicas the same question and keep the first answer."""
    winner = []

    async def ask(group, name, delay):
        answer = await call_backend(trace, name, delay, {"from": name})
        winner.append(answer)
        if hasattr(group, "cancel"):  # Python 3.15 and newer
            group.cancel()
        else:
            raise GotAnswer

    try:
        async with asyncio.TaskGroup() as group:
            group.create_task(ask(group, "replica-a", 0.30))
            group.create_task(ask(group, "replica-b", 0.10))
            group.create_task(ask(group, "replica-c", 0.20))
    except* GotAnswer:
        pass
    return winner[0]


async def main():
    trace = Trace()
    started = time.perf_counter()
    answer = await first_answer(trace)
    print(f"Python {sys.version.split()[0]}: {answer} after {time.perf_counter() - started:0.2f}s")
    trace.show()


asyncio.run(main())

On Python 3.13.14, which has no cancel():

Python 3.13.14: {'from': 'replica-b'} after 0.11s
  t=0.00s  replica-a: request sent
  t=0.00s  replica-b: request sent
  t=0.00s  replica-c: request sent
  t=0.11s  replica-b: answered
  t=0.11s  replica-b: connection released
  t=0.11s  replica-c: cancelled
  t=0.11s  replica-c: connection released
  t=0.11s  replica-a: cancelled
  t=0.11s  replica-a: connection released

On Python 3.15.0:

Python 3.15.0: {'from': 'replica-b'} after 0.12s
  t=0.00s  replica-a: request sent
  t=0.00s  replica-b: request sent
  t=0.00s  replica-c: request sent
  t=0.12s  replica-b: answered
  t=0.12s  replica-b: connection released
  t=0.12s  replica-c: cancelled
  t=0.12s  replica-c: connection released
  t=0.12s  replica-a: cancelled
  t=0.12s  replica-a: connection released

Both versions return replica B’s answer at about 0.1 seconds, and both cancel and release the other two replicas. On 3.15 no exception is involved at all, which makes the code shorter and easier to read once you can drop support for older versions.

Five mistakes to avoid

This last script shows five small mistakes in a row, so that you recognise their messages when they appear in your own code.

# step12_mistakes.py
import asyncio
import gc
import inspect
import warnings

from shop import ReviewsDown, Trace, reviews


async def create_after_exit():
    print("1. adding a task after the group has finished")
    async with asyncio.TaskGroup() as group:
        pass
    coroutine = reviews(Trace())
    try:
        group.create_task(coroutine)
    except RuntimeError as error:
        print(f"   RuntimeError: {error}")
    print(f"   state of the coroutine we passed in: {inspect.getcoroutinestate(coroutine)}")


async def result_too_early():
    print("2. asking for a result before the group has finished")
    async with asyncio.TaskGroup() as group:
        task = group.create_task(reviews(Trace()))
        try:
            task.result()
        except asyncio.InvalidStateError as error:
            print(f"   {type(error).__name__}: {error}")


async def body_error():
    print("3. an error raised in the body of the async with")
    try:
        async with asyncio.TaskGroup() as group:
            group.create_task(reviews(Trace()))
            raise ValueError("the body itself failed")
    except* ValueError as found:
        print(f"   it arrives inside the group: {[repr(e) for e in found.exceptions]}")


async def cancelled_siblings():
    print("4. siblings that were cancelled are not listed")
    try:
        async with asyncio.TaskGroup() as group:
            group.create_task(reviews(Trace(), delay=5))
            group.create_task(reviews(Trace(), delay=0.01, error=ReviewsDown("reviews returned 503")))
    except ExceptionGroup as found:
        print(f"   sub-exceptions: {[repr(e) for e in found.exceptions]}")


def forgot_await():
    print("5. calling a coroutine function without await")
    with warnings.catch_warnings(record=True) as caught:
        warnings.simplefilter("always")
        reviews(Trace())
        gc.collect()
    print(f"   {[str(item.message) for item in caught]}")


async def main():
    await create_after_exit()
    await result_too_early()
    await body_error()
    await cancelled_siblings()
    forgot_await()


asyncio.run(main())
1. adding a task after the group has finished
   RuntimeError: TaskGroup <TaskGroup entered> is finished
   state of the coroutine we passed in: CORO_CLOSED
2. asking for a result before the group has finished
   InvalidStateError: Result is not set.
3. an error raised in the body of the async with
   it arrives inside the group: ["ValueError('the body itself failed')"]
4. siblings that were cancelled are not listed
   sub-exceptions: ["ReviewsDown('reviews returned 503')"]
5. calling a coroutine function without await
   ["coroutine 'reviews' was never awaited"]
  1. Creating a task after the group has finished. The group raises RuntimeError, and it closes the coroutine you passed, so there is no “never awaited” warning on top. The documentation notes: “Changed in version 3.13: Close the given coroutine if the task group is not active.”
  2. Reading a result too early. Inside the block the task has not finished, so .result() raises InvalidStateError. Read results after the block.
  3. Expecting the body’s errors to stay separate. An error raised in the body arrives inside the group, next to any task failures, so the handler for it has to be an except* as well.
  4. Looking for cancelled siblings in the error. Tasks that were cancelled because of someone else’s failure are not listed in the group; only real failures are.
  5. Calling an async function without await. You get a coroutine object that never runs, and Python warns coroutine 'reviews' was never awaited.

One more rule applies when you start tasks outside a group. The create_task documentation warns: “The event loop only keeps weak references to tasks. A task that isn’t referenced elsewhere may get garbage collected at any time, even before it’s done.” A task group keeps its tasks alive for you, which is one more reason to use one.

Check that everything works end to end

From the taskgroups folder with the virtual environment active, run every script in order. The comments show what you should see; times will differ by a few hundredths on your machine.

python step00_versions.py          # three True lines
python step01_sequential.py        # about 0.60s one by one, about 0.30s together
python step02_gather_orphans.py    # 502 returned, then two answers arrive afterwards
python step02b_gather_values.py    # TypeError: 'ReviewsDown' object is not subscriptable
python step03_taskgroup.py         # 0.30s success; 0.06s failure, 0 tasks still running
python step04_except_star.py       # experiments A to F
python step04a_uncaught.py         # a traceback tree (exit code 1 is expected)
python step05_body_cancelled.py    # the body is interrupted at 0.06s
python step06_swallow.py           # buggy helper: about 0.27s; fixed: about 0.06s
python step06b_blocking.py         # about 0.5s blocking, about 0.06s in a thread
python step07_timeouts.py          # 0.26s with TimeoutError; 0.16s with except*
python step08_bounded.py           # 190 refused and peak 200; then 200 answers and peak 10
python step09_page_demo.py         # five scenarios
python -m pytest -q                # 9 passed
python step11_first_answer.py      # first answer wins
python step12_mistakes.py          # five messages

If a script says No module named 'shop', you are running it from a different folder than the one that holds shop.py. If step04_except_star.py fails to compile with a syntax error on the except* lines, you are on Python 3.10 or older. If a timing is far from the numbers above, close other heavy programs and run it again; the step 6.2 timings in particular show the same pattern at any speed.

Where this stops working, and what to learn next

A task group is a strong default, and it has limits that matter before you rely on it.

  • It is concurrency, not parallelism. Tasks take turns on one thread, so CPU-heavy functions still block the loop. A thread (asyncio.to_thread) helps for blocking input and output, and a separate process helps for heavy computation. Check the asyncio documentation for what threads can and cannot do in your Python build.
  • It is all or nothing by default. One failure cancels everything. If some calls are optional, wrap them as in optional from Step 9; never reach for except BaseException.
  • Cancellation is cooperative. A task stops only when it reaches an await. Blocking calls cannot be interrupted, and threads keep running (Step 6).
  • It does not retry. A failed call fails the group. Add a bounded retry with backoff inside the task, as in the retry tutorial, and stop calling a failing service for a while with a circuit breaker. If retries repeat a payment or another side effect, pair them with idempotency keys.
  • Shared work needs shielding. If several callers share one task, as in a cache, cancelling one caller must not cancel the work for the others. The cache stampede tutorial shows how asyncio.shield solves that.
  • It does not limit overall load. A semaphore limits one batch, not a whole service. For that, see the backpressure and load shedding tutorial.

To keep learning, read the asyncio task documentation for TaskGroup, timeout and cancellation, the language reference for except* and the exception group documentation, PEP 654 for the reasoning behind exception groups, the asyncio development guide for debug mode, and the synchronization primitives page for semaphores. You now have a working pattern, a test suite that pins down each behaviour, and a way to read the timeline of any concurrent code you write next.

Tags:

AsyncioConcurrencypytestPythonReliability EngineeringStructured Concurrency

Share

No Comment! Be the first one.

Leave a Reply

Your email address will not be published. Required fields are marked *

Related Posts