FastAPI AnyIO Structured Concurrency and TaskGroup Exception Aggregation Standard

--- description: FastAPI AnyIO Structured Concurrency and TaskGroup Exception Aggregation Standard - Enforces deterministic structured concurrency in FastAPI request lifecycles using AnyIO TaskGroups instead of unbounded fire-and-forget tasks or loose coroutine aggregation. Guarantees clean cancellation propagation to all child tasks upon client disconnect or downstream failure, preventing orphan background execution and resource leaks through unified ExceptionGroup handling. globs: app/**/*.py, src/**/*.py, routers/**/*.py, services/**/*.py, api/**/*.py alwaysApply: true verified_by: "agentrules.org" license: "CC-BY-4.0" source: "https://agentrules.org/rules/fastapi-anyio-structured-concurrency-taskgroup" --- # FastAPI AnyIO Structured Concurrency and TaskGroup Exception Aggregation Standard > **Severity**: CRITICAL | **Category**: backend > **Anti-Pattern Warning**: Spawning Unbounded Fire-and-Forget Coroutines and Swallowing Child Exceptions in FastAPI Request Cycles ## System Constraints You are a Principal Software Architect enforcing production-grade structured concurrency and resilience patterns in FastAPI applications. Mandate AnyIO TaskGroups and unified ExceptionGroup aggregation to eliminate orphan background tasks, socket leaks, and unhandled child coroutine failures during async request lifecycles. ## Strict Guidelines - Engineers MUST ALWAYS encapsulate concurrent child coroutines within an `anyio.create_task_group()` context manager instead of calling `asyncio.create_task()` directly within request-response cycles. - Engineers MUST NOT utilize `asyncio.gather(*tasks, return_exceptions=False)` without a structured parent context, as child failures leave siblings running as orphan coroutines. - Code MUST handle sub-task failures using Python 3.11+ `except*` syntax or `ExceptionGroup` filtering to guarantee that every concurrent exception is observed, logged, and mapped to an appropriate HTTP response. - Services MUST link child task cancellation to AnyIO `CancelScope` contexts or FastAPI request disconnection events, ensuring upstream client cancellations trigger immediate downstream resource cleanup. - Components NEVER execute long-running background processing inside request-scoped TaskGroups; true background operations must be dispatched to dedicated queue workers (e.g., Celery, ARQ) or lifespan-managed background task supervisors. ## Bad vs Good Implementations ### Concurrent API Call Orchestration in FastAPI #### Bad Antipattern (AI Common Hallucination) > **Why it fails**: Using bare `asyncio.create_task` and `asyncio.gather` does not guarantee cancellation of sibling tasks when one fails. The pending task continues running detached from the request lifecycle, leaking connections, CPU cycles, and memory. ```python import asyncio from fastapi import APIRouter, HTTPException router = APIRouter() async def fetch_user_data(user_id: str) -> dict: await asyncio.sleep(0.1) return {"id": user_id, "name": "Alice"} async def fetch_credit_score(user_id: str) -> dict: await asyncio.sleep(0.05) raise ConnectionResetError("Credit provider timed out") @router.get("/users/{user_id}/profile") async def get_user_profile(user_id: str): # ANTI-PATTERN: Bare create_task or gather leaves fetch_user_data running as an orphan # if fetch_credit_score raises an exception, and child exceptions are unhandled. task1 = asyncio.create_task(fetch_user_data(user_id)) task2 = asyncio.create_task(fetch_credit_score(user_id)) try: results = await asyncio.gather(task1, task2) return {"user": results[0], "credit": results[1]} except Exception as exc: # Orphaned task1 continues executing in the background, consuming resources raise HTTPException(status_code=502, detail=str(exc)) ``` #### Production Standard (Correct) > **Standard**: The AnyIO `TaskGroup` guarantees structured concurrency: if `_load_credit` raises an exception, the task group automatically cancels `_load_user` before exiting the scope. The Python 3.11+ `except*` block catches the aggregated `ExceptionGroup` deterministically. ```python import anyio from fastapi import APIRouter, HTTPException, status router = APIRouter() async def fetch_user_data(user_id: str) -> dict: await anyio.sleep(0.1) return {"id": user_id, "name": "Alice"} async def fetch_credit_score(user_id: str) -> dict: await anyio.sleep(0.05) raise ConnectionResetError("Credit provider timed out") @router.get("/users/{user_id}/profile") async def get_user_profile(user_id: str): user_profile = {} credit_profile = {} async def _load_user(): nonlocal user_profile user_profile = await fetch_user_data(user_id) async def _load_credit(): nonlocal credit_profile credit_profile = await fetch_credit_score(user_id) try: async with anyio.create_task_group() as tg: tg.start_soon(_load_user) tg.start_soon(_load_credit) except* ConnectionResetError as eg: # Downstream provider failure cleanly canceled all active siblings in tg raise HTTPException( status_code=status.HTTP_503_SERVICE_UNAVAILABLE, detail="External credit service unavailable" ) except* Exception as eg: raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"Failed to aggregate user profile: {eg.exceptions}" ) return {"user": user_profile, "credit": credit_profile} ```

Architectural Specification & Invariant Rationale

Architectural Invariant Design: FastAPI AnyIO Structured Concurrency and TaskGroup Exception Aggregation Standard

FastAPI AnyIO Structured Concurrency and TaskGroup Exception Aggregation Standard defines an industrial-grade behavioral standard for backend engineering. Enforces deterministic structured concurrency in FastAPI request lifecycles using AnyIO TaskGroups instead of unbounded fire-and-forget tasks or loose coroutine aggregation. Guarantees clean cancellation propagation to all child tasks upon client disconnect or downstream failure, preventing orphan background execution and resource leaks through unified ExceptionGroup handling. 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: Spawning Unbounded Fire-and-Forget Coroutines and Swallowing Child Exceptions in FastAPI Request Cycles 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	Spawning Unbounded Fire-and-Forget Coroutines and Swallowing Child Exceptions in FastAPI Request Cycles	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 concurrent child coroutines within an `anyio.create_task_group()` context manager instead of calling `asyncio.create_task()` directly within request-response cycles.

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.