Install & Compatibility
Where this runs
tested against v0.2.2 · pip install
no network on importno background threads
Install × environment matrix
Each cell = how many times install + import succeeded across repeated harness runs. Partial = flaky.
glibc = Debian/Ubuntu slim · musl = Alpine Linux
muslpy 3.10–3.95 runs
installs and imports cleanly · install 0.0s · import 0.240s · 18.3MB
glibcpy 3.10–3.95 runs
installs and imports cleanly · install 1.9s · import 0.202s · 19MB
16MB installed
● package 16MB
Code
Verified usage
Verified import paths — ran on the pinned version, not inferred.
TaskGroup
✓ from taskgroup import TaskGroup
For Python versions < 3.11, otherwise use `from asyncio import TaskGroup`.
Runner
✓ from taskgroup import Runner
For Python versions < 3.11, otherwise use `from asyncio import Runner`.
timeout
✓ from taskgroup import timeout
For Python versions < 3.11, otherwise use `from asyncio import timeout`.
run
✓ from taskgroup import run
For Python versions < 3.11, otherwise use `from asyncio import run`.
This quickstart demonstrates the usage of `TaskGroup`, `Runner`, and `timeout` through the `taskgroup` backport (or `asyncio` directly if Python 3.11+). It shows how to create and manage multiple concurrent tasks with `TaskGroup`, how to run a top-level coroutine with `Runner`, and how to apply a timeout to an asynchronous operation.
import sys
import asyncio
# Conditionally import TaskGroup, run, and timeout based on Python version
if sys.version_info >= (3, 11):
from asyncio import TaskGroup, run, timeout
else:
from taskgroup import TaskGroup, run, timeout
async def task_printer(name, delay):
print(f"Task {name}: Starting in {delay} seconds...")
await asyncio.sleep(delay)
print(f"Task {name}: Finished.")
return f"Result from {name}"
async def main_task_group():
print("Main: Creating tasks...")
async with TaskGroup() as group:
task1 = group.create_task(task_printer("A", 2))
task2 = group.create_task(task_printer("B", 1))
task3 = group.create_task(task_printer("C", 3))
print("Main: All tasks in group finished.")
print(f"Task A result: {task1.result()}")
print(f"Task B result: {task2.result()}")
print(f"Task C result: {task3.result()}")
async def main_runner():
print("\n--- Using Runner ---")
async def inner_coro():
await task_printer("D", 0.5)
await task_printer("E", 1.5)
with Runner() as runner:
runner.run(inner_coro())
print("Runner finished.")
async def main_timeout():
print("\n--- Using Timeout ---")
async def long_running_task():
await asyncio.sleep(5)
print("Long running task completed (should not happen if timeout works).")
try:
async with timeout(2):
await long_running_task()
except TimeoutError:
print("Task timed out as expected.")
if __name__ == "__main__":
print(f"Running with {'asyncio' if sys.version_info >= (3, 11) else 'taskgroup backport'}.")
run(main_task_group())
run(main_runner())
run(main_timeout())
Debug
Known issues
gotchaThis library is a backport. While it aims for faithful replication, subtle behavioral differences or edge cases might exist compared to the native `asyncio.TaskGroup`, `Runner`, or `timeout` in Python 3.11+. Always test thoroughly.fixRefer to the official `asyncio` documentation for `TaskGroup` behavior and be aware that the backport might not perfectly match all nuances, especially regarding complex cancellation scenarios or recent CPython bug fixes.
affects: < 0.2.2
breakingThe `asyncio.TaskGroup` requires Python 3.11 or higher. Using this backport on Python 3.11+ might lead to conflicts or unexpected behavior if both the native and backported versions are accidentally imported or used, although the provided quickstart handles this with a `sys.version_info` check.fixPrefer the native `asyncio.TaskGroup`, `Runner`, and `timeout` on Python 3.11 and later. Use conditional imports (as shown in the quickstart) to ensure the correct implementation is used for the host Python version.
affects: All versions when used on Python 3.11+
gotchaWhen an exception occurs in a task within an `asyncio.TaskGroup`, all other running tasks in that group are cancelled, and all non-cancellation exceptions are propagated as an `ExceptionGroup`. Proper error handling using `except*` (for `ExceptionGroup`) is crucial for robust applications. This behavior is fundamental to structured concurrency.fixFamiliarize yourself with `ExceptionGroup` and its handling via `except*` as introduced in Python 3.11. Design your async code within `TaskGroup` to gracefully handle task cancellations and aggregate exceptions.
affects: All versions
Errors
Common errors & fixes
ModuleNotFoundError: No module named 'taskgroup'
The `taskgroup` library has not been installed in the current Python environment.
ImportError: cannot import name 'TaskGroup' from 'asyncio'
On Python versions older than 3.11, `asyncio.TaskGroup` does not exist; users mistakenly try to import `TaskGroup` from the native `asyncio` module instead of the `taskgroup` backport.
fixfrom taskgroup import TaskGroup
AttributeError: module 'asyncio' has no attribute 'timeout'
Python versions older than 3.11 do not natively provide the `asyncio.timeout` context manager; users attempt to access it directly from `asyncio` instead of importing it from the `taskgroup` backport.
fixfrom taskgroup import timeout
Upgrade
Version history
0.2.2latest on PyPI · released Jan 3, 2025
Audit
Dependencies
No dependency data recorded yet.