FastAPI AnyIO Structured TaskGroup Exception Cancellation Standard

--- description: FastAPI AnyIO Structured TaskGroup Exception Cancellation Standard - Enforces strict structured concurrency using AnyIO TaskGroups for intra-request parallel coroutine execution within FastAPI endpoints. Guarantees deterministic cancellation cascading of sibling coroutines upon partial failure or client disconnection, preventing orphaned async workloads and database connection pool exhaustion. globs: app/**/*.py, src/**/*.py, api/**/*.py alwaysApply: true verified_by: "agentrules.org" license: "CC-BY-4.0" source: "https://agentrules.org/rules/fastapi-anyio-structured-taskgroup-cancellation-standard" --- # FastAPI AnyIO Structured TaskGroup Exception Cancellation Standard > **Severity**: CRITICAL | **Category**: backend > **Anti-Pattern Warning**: Unbounded asyncio.gather and Fire-and-Forget asyncio.create_task Leaking DB Sessions on Partial Request Failure ## System Constraints You are a Principal Backend Architect enforcing strict structured concurrency standards in FastAPI using AnyIO. Raw asyncio.gather, asyncio.create_task, and untracked background execution within request scopes are strictly prohibited. Mandate deterministic cancellation trees, explicit exception handling with ExceptionGroup handling, and strict resource scope lifecycles. ## Strict Guidelines - Engineers MUST ALWAYS use 'anyio.create_task_group()' for orchestrating concurrent sub-operations inside FastAPI request handlers. - Engineers MUST NOT invoke 'asyncio.create_task()' or 'asyncio.gather()' without structured cancellation boundaries inside request lifecycles. - Engineers MUST handle 'ExceptionGroup' (or BaseExceptionGroup in Python 3.11+) or catch specific inner exceptions when dealing with TaskGroup failures. - Services MUST ensure database sessions and network clients are managed outside individual child tasks or cleanly scoped such that task cancellation triggers deterministic cleanup and connection pool release. - Code SHOULD leverage AnyIO's cancellation tokens to gracefully terminate downstream HTTP or RPC calls when a sibling task fails. - Background work that outlives the HTTP request MUST be explicitly scheduled via FastAPI BackgroundTasks or dedicated worker queues (e.g., Celery, ARQ), NEVER implicitly orphaned in the request event loop. ## Bad vs Good Implementations ### Unstructured asyncio.gather Causing Orphaned Coroutines and Leaked Resources #### Bad Antipattern (AI Common Hallucination) > **Why it fails**: When 'charge_card' raises an exception, 'asyncio.gather' does not cancel 'reserve_inventory' by default unless explicit shield/cancellation handling is configured. The reserve task runs as a zombie, holding open the AsyncSession from the connection pool and potentially writing corrupted state. ```python import asyncio from fastapi import APIRouter, Depends, HTTPException from sqlalchemy.ext.asyncio import AsyncSession from app.db import get_db router = APIRouter() async def reserve_inventory(sku: str, db: AsyncSession): # Long-running DB write holding session lock await asyncio.sleep(2.0) # Writes to DB return True async def charge_card(amount: float): # Simulating sudden failure or validation crash await asyncio.sleep(0.5) raise ValueError("Payment gateway unreachable") @router.post("/checkout") async def checkout(sku: str, amount: float, db: AsyncSession = Depends(get_db)): try: # ANTI-PATTERN: If charge_card fails, reserve_inventory continues running # orphaned in the background, consuming connection pool resources and committing invalid state. results = await asyncio.gather( reserve_inventory(sku, db), charge_card(amount) ) return {"status": "success"} except Exception as exc: raise HTTPException(status_code=400, detail=str(exc)) ``` #### Production Standard (Correct) > **Standard**: Using 'anyio.create_task_group()' establishes strict structured concurrency. If 'charge_card' fails, AnyIO automatically triggers cancellation on the sibling '_safe_reserve' task, freeing the DB transaction and connection cleanly before bubbling up an ExceptionGroup. ```python import anyio from fastapi import APIRouter, Depends, HTTPException, status from sqlalchemy.ext.asyncio import AsyncSession from app.db import get_db router = APIRouter() async def reserve_inventory(sku: str, db: AsyncSession): # Correctly responsive to task group cancellation await anyio.sleep(2.0) return True async def charge_card(amount: float): await anyio.sleep(0.5) raise ValueError("Payment gateway unreachable") @router.post("/checkout") async def checkout(sku: str, amount: float, db: AsyncSession = Depends(get_db)): inventory_status = None async def _safe_reserve(): nonlocal inventory_status inventory_status = await reserve_inventory(sku, db) try: # AnyIO TaskGroup guarantees deterministic teardown: # If charge_card fails, the TaskGroup cancels _safe_reserve immediately. async with anyio.create_task_group() as tg: tg.start_soon(_safe_reserve) tg.start_soon(charge_card, amount) except* ValueError as eg: # Python 3.11+ except* syntax unwraps ExceptionGroup cleanly raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail=f"Checkout failed: {eg.exceptions[0]}" ) return {"status": "success", "reserved": inventory_status} ```

Architectural Specification & Invariant Rationale

Architectural Invariant Design: FastAPI AnyIO Structured TaskGroup Exception Cancellation Standard

FastAPI AnyIO Structured TaskGroup Exception Cancellation Standard defines an industrial-grade behavioral standard for backend engineering. Enforces strict structured concurrency using AnyIO TaskGroups for intra-request parallel coroutine execution within FastAPI endpoints. Guarantees deterministic cancellation cascading of sibling coroutines upon partial failure or client disconnection, preventing orphaned async workloads and database connection pool exhaustion. 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: Unbounded asyncio.gather and Fire-and-Forget asyncio.create_task Leaking DB Sessions on Partial Request Failure 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	Unbounded asyncio.gather and Fire-and-Forget asyncio.create_task Leaking DB Sessions on Partial Request Failure	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 use 'anyio.create_task_group()' for orchestrating concurrent sub-operations inside FastAPI request handlers.

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.