Python Basics: Finishing Async Work with TaskGroup and Timeouts

Related asynchronous operations should have a clear lifetime. asyncio.TaskGroup keeps their tasks inside one block, while asyncio.timeout bounds an awaitable operation and gives cleanup code a chance to run.

These examples use scheduling points and events instead of real network services. That keeps the results reproducible and makes cancellation visible without relying on a particular server, internet connection, or sleep duration.

Prepare a scratch directory

These three standalone examples were tested on Linux with Python 3.12.14. The examples require Python 3.11 or newer for asyncio.TaskGroup, asyncio.timeout, and except* exception-group handling. Use Bash, a text editor, and the standard mktemp, rm, and rmdir utilities. No third-party packages are required. Keep assertions enabled by running without Python’s -O option.

Create a fresh directory and keep using the same terminal:

start_dir="$PWD"
work_dir="$(mktemp -d)" || exit 1
cd -- "$work_dir" || exit 1

Save each named source file here as UTF-8.

Avoid source filenames such as asyncio.py. Run these as standalone terminal scripts, not inside an already-running notebook event loop. They do not create files or contact a service.

1. Collect a batch after its tasks finish

Save as collect_async_previews.py:

import asyncio


async def make_preview(name):
    await asyncio.sleep(0)
    return f"preview:{name}"


async def build_batch(names):
    async with asyncio.TaskGroup() as group:
        tasks = [group.create_task(make_preview(name)) for name in names]
    return [task.result() for task in tasks]


async def main():
    results = await build_batch(["cover", "chart"])
    empty = await build_batch([])
    assert results == ["preview:cover", "preview:chart"]
    assert empty == []
    print("Previews:", results)
    print("Empty batch:", empty)


asyncio.run(main())

Run it:

python3 collect_async_previews.py

Output:

Previews: ['preview:cover', 'preview:chart']
Empty batch: []

Leaving the TaskGroup waits for its tasks. Reading result after the block is therefore safe on this successful path. Keeping task handles in the original list also gives the output a predictable order, regardless of scheduling.

An empty batch works without a special branch. make_preview uses sleep(0) only to yield control; replace that stand-in with an actual asynchronous operation in an application. Declaring a function async does not by itself make blocking work cooperative.

2. Time out a wait and verify cleanup

Save as time_out_async_wait.py:

import asyncio


async def main():
    cleanup = []

    async def wait_for_reply():
        try:
            await asyncio.Event().wait()
        finally:
            cleanup.append("reply wait closed")

    try:
        async with asyncio.timeout(0):
            await wait_for_reply()
    except TimeoutError:
        print("Reply: timed out")
    else:
        raise AssertionError("A never-ready reply returned")
    assert cleanup == ["reply wait closed"]
    print("Cleanup:", cleanup[0])


asyncio.run(main())

Run it:

python3 time_out_async_wait.py

Output:

Reply: timed out
Cleanup: reply wait closed

The event is never set. A zero timeout schedules expiry on the next event-loop iteration, so this deliberately exercises the timeout path without measuring elapsed milliseconds. A real reply wait would normally use a positive deadline chosen for that operation.

The finally block runs before the outer handler prints the cleanup result. Catch TimeoutError outside the timeout context, where cancellation is converted into that exception. Keep cleanup short and reliable; the deadline is not a hard process-level kill switch.

3. Cancel sibling work when a batch fails

Save as cancel_failed_batch.py:

import asyncio


async def main():
    started = asyncio.Event()
    cleanup = []
    errors = []

    async def wait_for_asset():
        try:
            started.set()
            await asyncio.Event().wait()
        finally:
            cleanup.append("asset wait closed")

    async def reject_manifest():
        await started.wait()
        raise ValueError("manifest is missing a title")

    try:
        async with asyncio.TaskGroup() as group:
            asset = group.create_task(wait_for_asset())
            group.create_task(reject_manifest())
    except* ValueError as group_error:
        errors.extend(str(error) for error in group_error.exceptions)

    assert errors == ["manifest is missing a title"]
    assert asset.cancelled()
    assert cleanup == ["asset wait closed"]
    print("Batch error:", errors[0])
    print("Sibling cancelled:", asset.cancelled())
    print("Cleanup:", cleanup[0])


asyncio.run(main())

Run it:

python3 cancel_failed_batch.py

Output:

Batch error: manifest is missing a title
Sibling cancelled: True
Cleanup: asset wait closed

The started event ensures the asset wait has entered its cleanup-protected section before the manifest fails. TaskGroup cancels the unfinished sibling, waits for it, and reports the non-cancellation failure in an exception group.

except* handles the ValueError portion of that group. Other error types would remain unhandled rather than being mislabeled as a missing title. The checks confirm both cancellation and cleanup. The asset task does not swallow CancelledError; its finally block performs cleanup while cancellation continues normally.

Cooperative scheduling depends on returning control to the event loop. time.sleep, synchronous network calls, or a long calculation can delay every task and timeout in that loop. Choose asynchronous libraries or deliberately move suitable blocking work off the event-loop thread instead of expecting a TaskGroup to make it non-blocking.

Clean up

In the same terminal, remove only these source files, return to the original directory, and remove the now-empty scratch directory:

rm -- collect_async_previews.py time_out_async_wait.py cancel_failed_batch.py
cd -- "$start_dir" || exit 1
rmdir -- "$work_dir"
unset start_dir work_dir

If rmdir reports that anything remains, inspect the directory rather than deleting unknown files.

References

• Python 3.12 task groups
• asyncio timeouts
• Developing with asyncio: blocking code

Acknowledgements

This post was created with assistance from AI.