Tasks, Groups and Exception Groups

Warm-up: from Day 4

From memory first.

1. Predict both lines.

Warm-up: predict Python

class ORMError(Exception):
    pass

class NotFound(ORMError, LookupError):
    pass

try:
    try:
        {}["post"]
    except KeyError as error:
        raise NotFound("no post") from error
except LookupError as caught:
    print(type(caught).__name__, isinstance(caught, ORMError))
    print(type(caught.__cause__).__name__)

Check your prediction

NotFound True
KeyError

NotFound is both the library’s own error and a LookupError, so either kind of except catches it. raise ... from error keeps the original as __cause__. From An Exception Hierarchy of Your Own.

2. Fix. The caller must still hear about a failure after the rollback: this should print ['begin', 'rollback', 'caller saw it'].

Warm-up: fix Python

from contextlib import contextmanager

log = []

@contextmanager
def transaction():
    log.append("begin")
    try:
        yield
        log.append("commit")
    except Exception:
        log.append("rollback")

try:
    with transaction():
        raise ValueError("bad row")
except ValueError:
    log.append("caller saw it")

print(log)

Check your fix

from contextlib import contextmanager

log = []

@contextmanager
def transaction():
    log.append("begin")
    try:
        yield
        log.append("commit")
    except BaseException:
        log.append("rollback")
        raise

try:
    with transaction():
        raise ValueError("bad row")
except ValueError:
    log.append("caller saw it")

print(log)

['begin', 'rollback', 'caller saw it']

Handling an exception inside a context manager swallows it unless you re-raise. And BaseException, so a Ctrl-C rolls back too. From All or Nothing: Transactions.

3. Explain. No code: why should a retry loop catch only Retryable, and never Exception?

Check your explanation

Because a permanent bug — a TypeError from a typo — would then be retried several times and finally reported as “gave up”, disguising a bug in the code as a flaky network. Retry only what retrying can fix. From Testing the Paths That Fail.

Today

Today steps away from the ORM. Concurrency deserves its own day, and a library that does I/O has to answer questions the ORM never raises: how many things at once, what happens when one fails, and what is left running when a caller walks away.

We’ll work on simulated sources — asyncio.sleep standing in for a network — so every result here is reproducible.

A coroutine is not a task

Calling a coroutine function gives you an object that has not started:

Nothing happens until something schedules it Python

import asyncio


async def work(name):
    print(f"  {name} started")
    await asyncio.sleep(0.01)
    print(f"  {name} finished")
    return name


async def main():
    print("1. making a coroutine object")
    coroutine = work("first")          # nothing has run yet

    print("2. awaiting it")
    await coroutine                    # runs to completion, here and now

    print("3. creating a task")
    task = asyncio.create_task(work("second"))   # scheduled, not started
    print("   task created, still nothing printed above this line")
    await asyncio.sleep(0)             # one trip through the event loop
    print("4. after one tick, it has started")
    await task


asyncio.run(main())

await coroutine is sequential — it runs the work and waits for it. Concurrency starts when you create tasks, and a task doesn’t begin until the loop next gets control. Every await is such an opportunity; code between awaits runs without interruption.

That last point is the one worth keeping: async code is only concurrent at its await points. A “fast” function with no awaits blocks everything, and so does a slow synchronous call like time.sleep or a CPU-bound loop.

TaskGroup, not gather

asyncio.gather is the older way to run several things at once, and it has two behaviours you will eventually be bitten by: with return_exceptions=False a failure leaves the other tasks running, unobserved; with return_exceptions=True failures come back as values, which is easy to forget to check.

asyncio.TaskGroup (Python 3.11+) fixes both. When one task raises, the rest are cancelled, the group waits for them to finish cancelling, and it raises an ExceptionGroup.

One failure, and what happens to the siblings Python

import asyncio

finished = []


async def work(name, seconds, fail=False):
    try:
        await asyncio.sleep(seconds)
        if fail:
            raise RuntimeError(f"{name} failed")
        finished.append(name)
    except asyncio.CancelledError:
        finished.append(f"{name} cancelled")
        raise


async def with_group():
    finished.clear()
    try:
        async with asyncio.TaskGroup() as group:
            group.create_task(work("quick", 0.01))
            group.create_task(work("slow", 0.20))
            group.create_task(work("broken", 0.02, fail=True))
    except* RuntimeError as group_error:
        print("group raised:", [str(error) for error in group_error.exceptions])
    print("  finished:", finished)


async def with_gather():
    finished.clear()
    try:
        await asyncio.gather(
            work("quick", 0.01),
            work("slow", 0.20),
            work("broken", 0.02, fail=True),
        )
    except RuntimeError as error:
        print("gather raised:", error)
    await asyncio.sleep(0.25)          # wait and see what the siblings did
    print("  finished:", finished)


asyncio.run(with_group())
asyncio.run(with_gather())

With the group, slow is cancelled the moment broken fails, and the async with doesn’t return until that cancellation has been dealt with. With gather, the failure is reported immediately and slow carries on in the background — it finishes a fifth of a second later, into a program that has already moved on.

Structured concurrency

The rule a TaskGroup enforces is: no task outlives the block that started it. That’s what makes cleanup possible at all — if tasks can escape, there is no point at which you know everything has stopped. Use create_task directly only when you also own the cancelling, as run_pipeline does for its consumer.

Several failures at once

Concurrency means more than one thing can go wrong before you get a chance to react, so TaskGroup raises an ExceptionGroup — a single exception carrying several.

except* is how you handle them: each clause matches the exceptions of that type inside the group and leaves the rest to other clauses.

except* takes what it recognises Python

import asyncio


async def work(name, error=None):
    await asyncio.sleep(0.01)
    if error:
        raise error(f"{name}: {error.__name__}")
    return name


async def main():
    try:
        async with asyncio.TaskGroup() as group:
            group.create_task(work("a", ValueError))
            group.create_task(work("b", TimeoutError))
            group.create_task(work("c", ValueError))
            group.create_task(work("d"))
    except* ValueError as group_error:
        print("ValueErrors: ", [str(error) for error in group_error.exceptions])
    except* TimeoutError as group_error:
        print("TimeoutError:", [str(error) for error in group_error.exceptions])


asyncio.run(main())

Both clauses run: this is not if/elif. Two ValueErrors arrive together in one group, and the TimeoutError in another.

Three things catch people out:

except* clauses all run, in order, each receiving a sub-group containing only its matches.
A bare except ExceptionGroup also works, and then you unpack error.exceptions yourself. Fine for logging, worse for handling.
return is not allowed inside an except* block — it’s a SyntaxError. Set a flag or collect results and act after the try statement.

Exercise

Collect results and failures separately

gather_results should run every job concurrently and come back with both what worked and what didn’t — the shape a “fetch these ten things” API usually wants.

import asyncio
import unittest


class Failed(Exception):
    pass


async def job(name, fail=False, seconds=0.01):
    await asyncio.sleep(seconds)
    if fail:
        raise Failed(name)
    return f"{name}-result"


async def gather_results(names, failing=frozenset()):
    """Run every job concurrently.

    Return (results, failures): a dict of name -> result for the jobs that
    worked, and a dict of name -> exception for those that didn't. Nothing
    should be left running when this returns, and one failure must not stop
    the others.

    TODO: write it. A TaskGroup cancels siblings on failure, which is *not*
    what this function wants — think about where the try belongs.
    """
    raise NotImplementedError("gather_results")


class Gathering(unittest.TestCase):
    def test_all_succeed(self):
        results, failures = asyncio.run(gather_results(["a", "b"]))
        self.assertEqual(results, {"a": "a-result", "b": "b-result"})
        self.assertEqual(failures, {})

    def test_failures_are_reported_separately(self):
        results, failures = asyncio.run(
            gather_results(["a", "b", "c"], failing=frozenset({"b"}))
        )
        self.assertEqual(sorted(results), ["a", "c"])
        self.assertIsInstance(failures["b"], Failed)

    def test_a_failure_does_not_cancel_the_others(self):
        results, failures = asyncio.run(
            gather_results(["fast", "slow"], failing=frozenset({"fast"}))
        )
        self.assertIn("slow", results)

    def test_nothing_is_left_running(self):
        async def scenario():
            await gather_results(["a", "b"], failing=frozenset({"a"}))
            return [task for task in asyncio.all_tasks() if not task.done()]

        self.assertEqual(len(asyncio.run(scenario())), 1)


unittest.TextTestRunner(verbosity=2).run(
    unittest.defaultTestLoader.loadTestsFromTestCase(Gathering)
)

Hint 1: the idea

A TaskGroup cancels every sibling as soon as one task fails — the opposite of what’s wanted here. So catch each job’s expected failure inside the task that runs it, and record it there. Then the group never sees a failure, and every job gets to finish.

Hint 2: a first step

Write a small inner async def run(name) that awaits the job inside a try, putting the result in one dictionary or the error in the other. Then start one run per name inside async with asyncio.TaskGroup() as group:.

Show one solution

async def gather_results(names, failing=frozenset()):
    results, failures = {}, {}

    async def run(name):
        try:
            results[name] = await job(name, fail=name in failing)
        except Failed as error:
            failures[name] = error          # caught here, so the group never sees it

    async with asyncio.TaskGroup() as group:
        for name in names:
            group.create_task(run(name))
    return results, failures

The trick is where the try goes. Inside the worker, a failure is handled before it can reach the group, so no sibling is cancelled — you keep the group’s guarantee that nothing outlives the block, without its all-or-nothing behaviour.

Wrapping the async with in a try instead would give you the opposite: siblings cancelled, and an ExceptionGroup to unpick. That’s the right choice when a failure means the whole operation is pointless, which is what read_all_strict does in today’s library.

Writing to results and failures from several tasks needs no lock: tasks only switch at await, and neither assignment contains one. That is a real guarantee of the single-threaded event loop, and one of the few places where async is genuinely simpler than threads.

What you learned

A coroutine object does nothing until awaited or wrapped in a task; tasks start at the next trip through the loop.
Async code interleaves only at await, so a blocking call stops everything.
TaskGroup cancels siblings on failure and waits for them — no task outlives the block.
ExceptionGroup and except* report and handle several failures at once, and every matching clause runs.