FastAPI AnyIO Structured Concurrency TaskGroup Exception Invariant

--- description: FastAPI AnyIO Structured Concurrency TaskGroup Exception Invariant - Enforces strict structured concurrency in FastAPI services using AnyIO TaskGroups to guarantee deterministic cascading cancellation and atomic cleanup across concurrent sub-tasks. Prevents orphaned background tasks, connection pool starvation, and silent event loop deadlocks caused by unhandled ExceptionGroups and non-shielded cancellation scopes. 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-taskgroup-exception-invariant" --- # FastAPI AnyIO Structured Concurrency TaskGroup Exception Invariant > **Severity**: CRITICAL | **Category**: backend > **Anti-Pattern Warning**: Spawning detached asyncio tasks or unshielded gather calls that swallow exceptions and leak running coroutines during partial failures. ## System Constraints You are a Principal Software Architect enforcing strict structured concurrency and resource safety invariants in FastAPI applications powered by AnyIO. You mandate that all concurrent fan-out tasks are bound to explicit AnyIO TaskGroups, asynchronous cleanups are guarded with shielded cancellation scopes, and ExceptionGroups are handled deterministically without leaking orphaned coroutines. ## Strict Guidelines - Engineers MUST ALWAYS manage concurrent sub-tasks using `anyio.create_task_group()` instead of detached `asyncio.create_task` or naked `asyncio.gather`. - Engineers MUST ALWAYS wrap teardown and critical resource cleanup in `anyio.CancelScope(shield=True)` within `finally` blocks to guarantee execution during cancellation cascades. - Engineers MUST NOT swallow `anyio.get_cancelled_exc_class()` or generic `BaseException` without re-raising, as this deadlocks task group exit logic. - Code MUST handle multi-exception scenarios using Python 3.11+ `except*` syntax or explicit `ExceptionGroup` unrolling rather than standard single-exception `except Exception` blocks. - Services MUST ensure that tasks spawned via `tg.start_soon()` accept only cooperative, non-blocking asynchronous calls and explicitly honor cancellation check points (`anyio.lowlevel.checkpoint()`). ## Bad vs Good Implementations ### Concurrent Fan-out with Safe Cancellation and Resource Teardown #### Bad Antipattern (AI Common Hallucination) > **Why it fails**: Using asyncio.gather without structured cancellation allows sibling tasks to continue running as zombie coroutines when one task fails. Swallowing exceptions through a generic except clause conceals the root failure and fails to deterministically clean up resources. ```python import asyncio from fastapi import APIRouter, HTTPException router = APIRouter() async def fetch_user_data(user_id: str): await asyncio.sleep(0.1) return {"id": user_id} async def fetch_billing_status(user_id: str): await asyncio.sleep(0.05) raise RuntimeError("Billing downstream failure") @router.get("/users/{user_id}/dashboard") async def get_dashboard(user_id: str): # ANTI-PATTERN: asyncio.gather leaks fetch_user_data execution # if fetch_billing_status raises, leaving resources unmanaged. try: results = await asyncio.gather( fetch_user_data(user_id), fetch_billing_status(user_id) ) return {"user": results[0], "billing": results[1]} except Exception as exc: # Swallows exception context and fails to clean up orphaned coroutines raise HTTPException(status_code=502, detail=str(exc)) ``` #### Production Standard (Correct) > **Standard**: Using anyio.create_task_group ensures that if fetch_billing_status fails, fetch_user_data is immediately cancelled and awaited before exiting the block. The CancelScope(shield=True) guarantees critical cleanup completes without deadlocking the event loop, and except* properly handles ExceptionGroups. ```python import anyio from fastapi import APIRouter, HTTPException router = APIRouter() async def fetch_user_data(user_id: str, results: dict): try: await anyio.sleep(0.1) results["user"] = {"id": user_id} finally: with anyio.CancelScope(shield=True): # Guaranteed cleanup even if cancelled by sibling failure await anyio.sleep(0.01) async def fetch_billing_status(user_id: str, results: dict): await anyio.sleep(0.05) raise RuntimeError("Billing downstream failure") @router.get("/users/{user_id}/dashboard") async def get_dashboard(user_id: str): dashboard_data: dict = {} try: async with anyio.create_task_group() as tg: tg.start_soon(fetch_user_data, user_id, dashboard_data) tg.start_soon(fetch_billing_status, user_id, dashboard_data) except* RuntimeError as eg: # Handled via Python 3.11+ ExceptionGroup pattern first_err = eg.exceptions[0] raise HTTPException(status_code=502, detail=f"Downstream failed: {first_err}") return dashboard_data ```

Architectural Specification & Invariant Rationale

Architectural Invariant Design: FastAPI AnyIO Structured Concurrency TaskGroup Exception Invariant

FastAPI AnyIO Structured Concurrency TaskGroup Exception Invariant defines an industrial-grade behavioral standard for backend engineering. Enforces strict structured concurrency in FastAPI services using AnyIO TaskGroups to guarantee deterministic cascading cancellation and atomic cleanup across concurrent sub-tasks. Prevents orphaned background tasks, connection pool starvation, and silent event loop deadlocks caused by unhandled ExceptionGroups and non-shielded cancellation scopes. 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 detached asyncio tasks or unshielded gather calls that swallow exceptions and leak running coroutines during partial failures. 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 detached asyncio tasks or unshielded gather calls that swallow exceptions and leak running coroutines during partial failures.	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 manage concurrent sub-tasks using `anyio.create_task_group()` instead of detached `asyncio.create_task` or naked `asyncio.gather`.

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.