FastAPI AnyIO Structured Concurrency and ExceptionGroup Invariant

--- description: FastAPI AnyIO Structured Concurrency and ExceptionGroup Invariant - Enforces strict structured concurrency in FastAPI ASGI event loops using AnyIO task groups and Python 3.11+ ExceptionGroup semantics. Prevents orphaned coroutines, unhandled task cancellation leaks, and dangling socket descriptors caused by improper concurrent fan-out or raw asyncio primitives. 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-concurrency-exceptiongroup-invariant" --- # FastAPI AnyIO Structured Concurrency and ExceptionGroup Invariant > **Severity**: CRITICAL | **Category**: backend > **Anti-Pattern Warning**: Unstructured asyncio.gather or BackgroundTasks Masking Concurrency Exceptions and Leaking Sockets ## System Constraints You are a Principal Backend Architect enforcing structured concurrency and deterministic lifecycle management in FastAPI services running on AnyIO. You must reject any code using unstructured concurrency primitives (such as raw asyncio.create_task or fire-and-forget asyncio.gather) and mandate anyio.create_task_group() with explicit ExceptionGroup handling (via except* in Python 3.11+) to guarantee cancellation tree propagation and prevent socket resource leaks. ## Strict Guidelines - Engineers MUST ALWAYS use `anyio.create_task_group()` context managers for concurrent I/O fan-out within FastAPI endpoints and service layers. - Engineers MUST NOT invoke `asyncio.create_task`, `asyncio.gather`, or `asyncio.shield` directly within AnyIO-managed ASGI request cycles, as they break structured cancellation guarantees. - Services MUST catch aggregate multi-failure exceptions using Python 3.11+ `except*` syntax or `exceptiongroup.catch` to prevent unhandled sub-exceptions from terminating the ASGI worker. - Engineers MUST ensure task cancellation tokens propagate cleanly by never suppressing `anyio.get_cancelled_exc_class()` within broad `except Exception:` blocks. - Background worker operations spawned from request handlers MUST yield to deterministic task scopes or explicit worker pools rather than untracked fire-and-forget coroutine tasks. ## Bad vs Good Implementations ### Concurrent Fan-Out with Exception Handling and Resource Clean-up #### Bad Antipattern (AI Common Hallucination) > **Why it fails**: Using asyncio.gather inside FastAPI allows the surviving coroutine to run detached if one raises an exception, leaving active HTTP sockets and database connections orphaned in the ASGI event loop. ```python import asyncio from fastapi import APIRouter, HTTPException router = APIRouter() async def fetch_upstream_a(client): return await client.get("https://service-a/data") async def fetch_upstream_b(client): return await client.get("https://service-b/data") @router.get("/aggregate") async def aggregate_data(): # ANTI-PATTERN: asyncio.gather leaks background tasks if one fails # and fails to propagate structured cancellation to open sockets. try: res_a, res_b = await asyncio.gather( fetch_upstream_a(None), fetch_upstream_b(None) ) return {"a": res_a, "b": res_b} except Exception as exc: # Other pending tasks continue running orphaned in the event loop raise HTTPException(status_code=502, detail=str(exc)) ``` #### Production Standard (Correct) > **Standard**: anyio.create_task_group ensures that if fetch_upstream_b fails, fetch_upstream_a is immediately and deterministically cancelled, sockets are finalized, and multi-exceptions are cleanly handled via except*. ```python import anyio from fastapi import APIRouter, HTTPException router = APIRouter() async def fetch_upstream_a(results: dict): # Simulated I/O results["a"] = {"status": "ok"} async def fetch_upstream_b(results: dict): # Simulated I/O failure raise ConnectionResetError("Upstream peer disconnected") @router.get("/aggregate") async def aggregate_data(): results = {} try: # PRODUCTION INVARIANT: Structured concurrency via AnyIO TaskGroup async with anyio.create_task_group() as tg: tg.start_soon(fetch_upstream_a, results) tg.start_soon(fetch_upstream_b, results) except* ConnectionResetError as eg: # Sub-tree cancellation is guaranteed; sibling tasks are reliably cancelled raise HTTPException(status_code=502, detail=f"Upstream failure: {eg.exceptions}") except* Exception as eg: raise HTTPException(status_code=500, detail="Internal concurrent failure") return results ```

Architectural Specification & Invariant Rationale

Architectural Invariant Design: FastAPI AnyIO Structured Concurrency and ExceptionGroup Invariant

FastAPI AnyIO Structured Concurrency and ExceptionGroup Invariant defines an industrial-grade behavioral standard for backend engineering. Enforces strict structured concurrency in FastAPI ASGI event loops using AnyIO task groups and Python 3.11+ ExceptionGroup semantics. Prevents orphaned coroutines, unhandled task cancellation leaks, and dangling socket descriptors caused by improper concurrent fan-out or raw asyncio primitives. 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: Unstructured asyncio.gather or BackgroundTasks Masking Concurrency Exceptions and Leaking Sockets 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	Unstructured asyncio.gather or BackgroundTasks Masking Concurrency Exceptions and Leaking Sockets	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()` context managers for concurrent I/O fan-out within FastAPI endpoints and service layers.

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.