FastAPI AnyIO Structured TaskGroup & ExceptionGroup Propagation Standard

--- description: FastAPI AnyIO Structured TaskGroup & ExceptionGroup Propagation Standard - Enforces structured concurrency across asynchronous FastAPI endpoints and background routines using AnyIO TaskGroups and Python 3.11+ ExceptionGroups. Guarantees cooperative cancellation across concurrent subtasks when any subtask faults, eliminating orphaned processes, dangling database sessions, and silent background resource leaks. globs: app/**/*.py, src/**/*.py, services/**/*.py, workers/**/*.py alwaysApply: true verified_by: "agentrules.org" license: "CC-BY-4.0" source: "https://agentrules.org/rules/fastapi-anyio-taskgroup-cancellation-standard" --- # FastAPI AnyIO Structured TaskGroup & ExceptionGroup Propagation Standard > **Severity**: CRITICAL | **Category**: backend > **Anti-Pattern Warning**: Unsupervised asyncio.create_task Spawning and Bare Exception Catching Causing Orphaned Worker Proliferation ## System Constraints You are a Principal Software Architect enforcing the FastAPI AnyIO Structured TaskGroup ExceptionGroup Propagation Standard. You must forbid unmanaged concurrency via bare asyncio.create_task, raw BackgroundTasks without lifecycle management, and generic Exception swallows. Require all co-dependent async operations to execute inside an AnyIO TaskGroup context, use Python 3.11+ except* syntax for ExceptionGroup handling, and guarantee that partial service failures trigger immediate cooperative cancellation of sibling tasks. ## Strict Guidelines - Engineers MUST ALWAYS encapsulate co-dependent asynchronous background operations within `async with anyio.create_task_group() as tg:` rather than invoking `asyncio.create_task` directly. - Engineers MUST NOT swallow `BaseExceptionGroup` or `ExceptionGroup` with bare `except Exception:` blocks, as this suppresses cooperative cancellation signals and masks concurrent failures. - Engineers MUST use Python 3.11+ `except*` syntax (or AnyIO's `MultiError` catchers on legacy runtimes) to inspect and handle granular exception types within propagated `ExceptionGroup` boundaries. - Services MUST explicitly shield resource cleanups using `with anyio.CancelScope(shield=True):` to ensure rollbacks, socket closures, or state updates complete even during parent task cancellation. - Components NEVER start long-lived background tasks decoupled from request or application lifespans without an explicit registered supervisor or cancellation token. ## Bad vs Good Implementations ### Unstructured asyncio.create_task vs AnyIO Structured TaskGroup with ExceptionGroup Handling #### Bad Antipattern (AI Common Hallucination) > **Why it fails**: Using asyncio.create_task with asyncio.gather without strict cancellation semantics causes process_metadata to run as a zombie task after process_media fails. Catching generic Exception masks partial concurrency failures, leaks system resources, and corrupts backend persistence layers. ```python import asyncio from fastapi import APIRouter, HTTPException router = APIRouter() async def process_media(upload_id: str): await asyncio.sleep(2) raise ConnectionError("Storage backend unavailable") async def process_metadata(upload_id: str): # Runs indefinitely if process_media fails, corrupting state for i in range(10): await asyncio.sleep(1) @router.post("/upload/{upload_id}") async def upload_asset(upload_id: str): try: # ANTI-PATTERN: Fire-and-forget concurrent tasks decouple cancellation. # If one fails, the other becomes an orphaned zombie worker. t1 = asyncio.create_task(process_media(upload_id)) t2 = asyncio.create_task(process_metadata(upload_id)) await asyncio.gather(t1, t2) except Exception as e: # Swallows raw exceptions, leaves process_metadata running raise HTTPException(status_code=500, detail=str(e)) ``` #### Production Standard (Correct) > **Standard**: Enclosing concurrent tasks in `anyio.create_task_group()` ensures that when `process_media` raises `ConnectionError`, the enclosing scope automatically cancels `process_metadata`. Critical cleanup is secured via `CancelScope(shield=True)`, and Python 3.11+ `except*` routes the `ExceptionGroup` appropriately without leaking processes. ```python import anyio from fastapi import APIRouter, HTTPException router = APIRouter() async def process_media(upload_id: str): await anyio.sleep(2) raise ConnectionError("Storage backend unavailable") async def process_metadata(upload_id: str): try: for _ in range(10): await anyio.sleep(1) finally: # INVARIANT: Shield critical rollback logic from cancellation cascades with anyio.CancelScope(shield=True): # Perform idempotent cleanup pass @router.post("/upload/{upload_id}") async def upload_asset(upload_id: str): try: # INVARIANT: Structured TaskGroup ensures immediate sibling cancellation upon any failure async with anyio.create_task_group() as tg: tg.start_soon(process_media, upload_id) tg.start_soon(process_metadata, upload_id) except* ConnectionError as eg: # Python 3.11+ granular ExceptionGroup handling raise HTTPException(status_code=503, detail="Upstream storage failure during processing") except* Exception as eg: raise HTTPException(status_code=500, detail="Internal task group failure") ```

Architectural Specification & Invariant Rationale

Architectural Invariant Design: FastAPI AnyIO Structured TaskGroup & ExceptionGroup Propagation Standard

FastAPI AnyIO Structured TaskGroup & ExceptionGroup Propagation Standard defines an industrial-grade behavioral standard for backend engineering. Enforces structured concurrency across asynchronous FastAPI endpoints and background routines using AnyIO TaskGroups and Python 3.11+ ExceptionGroups. Guarantees cooperative cancellation across concurrent subtasks when any subtask faults, eliminating orphaned processes, dangling database sessions, and silent background resource leaks. Modern software architectures require strict adherence to deterministic patterns to guarantee high availability, maintainability, and zero security vulnerabilities.

Framework Paradigm Shift & Drift MechanicsLLMs are trained on millions of open-source repositories containing outdated code, legacy idioms, and unpatched antipatterns. When coding assistants generate implementations for this stack, they default to common statistical tokens: Unsupervised asyncio.create_task Spawning and Bare Exception Catching Causing Orphaned Worker Proliferation Without an explicit system instruction file, developers spend hours debugging silent failures.

Failure Modes & Production Blast Radius

Hazard Analysis: What Breaks Without This Rule

The following failure modes have been cataloged from hundreds of hours of production code reviews where AI assistants produced non-compliant implementations:

Failure Mode	AI Symptom (The Anti-Pattern)	Severity	Production Impact & Remedy
AI Hallucination & Deprecation Drift	Unsupervised asyncio.create_task Spawning and Bare Exception Catching Causing Orphaned Worker Proliferation	Critical	Violating this industrial standard introduces regression bugs, security risks, and unexpected runtime failures in production. Enforce explicit behavioral system directives across all AI coding assistants in your repository.
Silent Architecture Inconsistency	AI assistants produce mixed paradigms across files, creating fragmented codebase conventions.	Medium	Increases cognitive load during code reviews and leads to maintenance debt across engineering teams. Pin project conventions in .cursor/rules, CLAUDE.md, and AGENTS.md so all team agents follow identical guardrails.

Architectural Directives in Detail

Step-by-Step Production Directives

Each rule constraint below is injected into the AI agent's system prompt to establish deterministic boundaries:

Engineers MUST ALWAYS encapsulate co-dependent asynchronous background operations within `async with anyio.create_task_group() as tg

Engineers MUST ALWAYS encapsulate co-dependent asynchronous background operations within `async with anyio.create_task_group() as tg:` rather than invoking `asyncio.create_task` directly.

Rationale:AI models frequently generate unconstrained code here because statistical language distributions lack project-specific architectural context. Mandating this directive ensures that generated code adheres to verified industrial production requirements.