Skip to content
DevOps AI ToolKit
Newsletter
All prompts
AI for Bash & Python Automation Difficulty: Advanced ClaudeChatGPTCursor

Asyncio TaskGroup Structured-Concurrency Python Script Prompt

Write a Python 3.11+ automation script that fans work out with asyncio.TaskGroup so failures cancel siblings cleanly, errors aggregate into an ExceptionGroup, and no task is ever silently orphaned.

Target user
Python automation engineers and SREs writing concurrent I/O jobs
Difficulty
Advanced
Tools
Claude, ChatGPT, Cursor

The prompt

You are a senior Python engineer who writes correct async automation. I have a concurrent job that fans out I/O-bound work (HTTP calls, DB queries, subprocess launches) and I keep hitting the classic `asyncio.gather` traps: a failing task leaves siblings running, exceptions get swallowed, and Ctrl-C leaves "Task was destroyed but it is pending" warnings. I want to rewrite it with `asyncio.TaskGroup` (Python 3.11+) so concurrency is structured and failures are deterministic.

I will provide:
- The unit of work (a coroutine) and the collection of inputs to run it over
- The concurrency ceiling (max in-flight), timeouts, and whether partial success is acceptable
- Python version and whether a 3.10 fallback is needed

Your job:

1. **Model the fan-out with `async with asyncio.TaskGroup() as tg`** — create tasks via `tg.create_task(...)`, and explain the core guarantee: when the block exits, every task has completed or the group has cancelled the rest and re-raised.
2. **Bound concurrency** — pair the TaskGroup with an `asyncio.Semaphore` so only N coroutines run at once; show the `async with sem:` wrapper inside the worker, not around task creation.
3. **Aggregate errors correctly** — a TaskGroup raises an `ExceptionGroup`; show how to catch it with `except* SomeError:` (3.11 syntax) and how to separate "cancel everything on first failure" from "collect all results and report failures at the end" (return exceptions as values when partial success is fine).
4. **Add per-task timeouts** — use `asyncio.timeout()` inside each worker so one slow task can't stall the group, and make the timeout error carry which input it came from.
5. **Handle cancellation cleanly** — ensure Ctrl-C / SIGINT cancels the group, awaits teardown, and exits without pending-task warnings; note the re-raise of `CancelledError`.
6. **Provide the 3.10 fallback** — show the equivalent `asyncio.gather(..., return_exceptions=True)` pattern and call out exactly what safety you lose versus TaskGroup.

Output: the complete typed script with a runnable `asyncio.run(main())` entrypoint, a table comparing `gather` vs `TaskGroup` semantics, and two tests — one where a mid-flight task raises (proving siblings cancel) and one all-success run.

Run this prompt with AI

Test it, get an AI-improved version, or compare models — live in the Prompt Workspace. No copy-paste.

Related prompts

More Bash & Python Automation prompts & error guides

Browse every Bash & Python Automation prompt and troubleshooting guide in one place.

Free download · 368-page PDF

Reading prompts? Get all 500 in one free PDF

500 battle-tested, copy-paste AI prompts engineered by a senior systems engineer — every one with fill-in placeholders and safety/back-out notes. Drop your email and it's yours.

  • 500 prompts: Linux · Kubernetes · Terraform · OpenStack · GitLab · Docker · Monitoring · Incident Response
  • Instant PDF download — yours free, forever
  • Plus one practical AI-workflow email a week (no spam)

Single opt-in · unsubscribe anytime · no spam.