中文
devexamples

Python asyncio TaskGroup Batch Runner

Run coroutines in an asyncio.TaskGroup so the first failure cancels siblings, then inspect the raised ExceptionGroup.

PythonResilience

Code

snippet
import asyncio
from dataclasses import dataclass


@dataclass
class Job:
    name: str
    seconds: float


async def run_job(job: Job) -> tuple[str, str]:
    await asyncio.sleep(job.seconds)
    if job.name == "bad":
        raise RuntimeError(f"job {job.name} failed")
    return job.name, "ok"


async def run_batch(jobs: list[Job]) -> dict[str, str]:
    tasks: list[tuple[str, asyncio.Task[tuple[str, str]]]] = []
    async with asyncio.TaskGroup() as group:
        for job in jobs:
            tasks.append((job.name, group.create_task(run_job(job))))
    return {name: task.result()[1] for name, task in tasks}


async def main() -> None:
    print(await run_batch([Job("alpha", 0.05), Job("beta", 0.01)]))
    try:
        await run_batch([Job("slow", 1.0), Job("bad", 0.01)])
    except* RuntimeError as group:
        print("failures:", [str(error) for error in group.exceptions])
        print("slow job cancelled by TaskGroup")


asyncio.run(main())

Related Examples