FastAPI with AnyIO 4 Structured Concurrency TaskGroup Invariant

--- description: FastAPI with AnyIO 4 Structured Concurrency TaskGroup Invariant - Enforces strict structured concurrency invariants using AnyIO 4 TaskGroups across FastAPI ASGI endpoints and service boundaries. Prevents orphaned background tasks, broken cancellation cascades, and uncontrolled exception fan-out across concurrent external I/O operations by mandating deterministic cancellation scopes and PEP 654 ExceptionGroup handling. globs: app/**/*.py, src/**/*.py, api/**/*.py alwaysApply: true verified_by: "agentrules.org" license: "CC-BY-4.0" source: "https://agentrules.org/rules/fastapi-anyio4-structured-concurrency-invariant" --- # FastAPI with AnyIO 4 Structured Concurrency TaskGroup Invariant > **Severity**: CRITICAL | **Category**: backend > **Anti-Pattern Warning**: Unbounded asyncio.create_task and asyncio.gather Fan-Out Leaking Coroutines Across ASGI Worker Lifecycles ## System Constraints You are a Principal Software Architect enforcing AnyIO 4 structured concurrency invariants in FastAPI applications. You must forbid naked asyncio concurrency primitives (such as asyncio.create_task, asyncio.gather, or asyncio.shield) within ASGI request-response lifecycles. Mandate the use of AnyIO 4 TaskGroups with explicit cancellation deadlines, robust PEP 654 ExceptionGroup isolation, and deterministic resource lifespans to guarantee that no asynchronous child coroutine can outlive or silently fail outside its parent scope. ## Strict Guidelines - Engineers MUST ALWAYS encapsulate concurrent asynchronous I/O operations within an 'async with anyio.create_task_group() as tg:' context block. - Engineers MUST NOT invoke 'asyncio.create_task()', 'asyncio.gather()', or 'asyncio.ensure_future()' within request-scoped route handlers or dependency lifecycles. - Code SHOULD wrap AnyIO TaskGroups within explicit timeout boundaries using 'anyio.fail_after()' or 'anyio.move_on_after()' to enforce deterministic termination. - Services MUST catch and process concurrent failures using PEP 654 'except*' syntax or explicit AnyIO ExceptionGroup unwrapping to prevent unhandled tracebacks leaking through ASGI middleware. - Components NEVER allow child coroutines to outlive the enclosing ASGI request context scope unless dispatched via managed worker queues or lifespan background runners. ## Bad vs Good Implementations ### Concurrent Fan-Out with Deterministic Cancellation and Exception Isolation #### Bad Antipattern (AI Common Hallucination) > **Why it fails**: Using asyncio.create_task and asyncio.gather does not guarantee automatic cancellation of sibling tasks when one fails. The failing task leaves the remaining task executing as an orphan, which consumes ASGI worker connections and can mutate data out-of-band. ```python import asyncio from fastapi import APIRouter, HTTPException router = APIRouter() @router.get("/user-dashboard/{user_id}") async def get_dashboard(user_id: str): # Anti-pattern: Naked asyncio.gather with fire-and-forget task creation # If fetch_notifications fails, fetch_metrics continues running orphaned in background task_metrics = asyncio.create_task(fetch_metrics(user_id)) task_notifications = asyncio.create_task(fetch_notifications(user_id)) try: metrics, notifications = await asyncio.gather(task_metrics, task_notifications) return {"metrics": metrics, "notifications": notifications} except Exception as exc: # Orphaned tasks keep running, leaking sockets and memory raise HTTPException(status_code=500, detail="Dashboard fetch failed") ``` #### Production Standard (Correct) > **Standard**: AnyIO 4 TaskGroup guarantees structured concurrency: if any child task raises an exception, the task group automatically cancels all other running child tasks before re-raising. Encapsulating this inside anyio.fail_after guarantees deterministic timeouts, and PEP 654 except* enables precise multi-exception handling. ```python import anyio from fastapi import APIRouter, HTTPException router = APIRouter() @router.get("/user-dashboard/{user_id}") async def get_dashboard(user_id: str): metrics: dict | None = None notifications: list | None = None async def _fetch_metrics_task() -> None: nonlocal metrics metrics = await fetch_metrics(user_id) async def _fetch_notifications_task() -> None: nonlocal notifications notifications = await fetch_notifications(user_id) try: # Enforce deterministic deadline and structured cancellation fan-out async with anyio.fail_after(2.5): async with anyio.create_task_group() as tg: tg.start_soon(_fetch_metrics_task) tg.start_soon(_fetch_notifications_task) except* TimeoutError: raise HTTPException(status_code=504, detail="Upstream aggregation timed out") except* UpstreamDependencyError as eg: raise HTTPException(status_code=502, detail=f"Upstream failure: {eg.exceptions[0]}") return {"metrics": metrics, "notifications": notifications} ```

Architectural Specification & Invariant Rationale

Architectural Invariant Design: FastAPI with AnyIO 4 Structured Concurrency TaskGroup Invariant

FastAPI with AnyIO 4 Structured Concurrency TaskGroup Invariant defines an industrial-grade behavioral standard for backend engineering. Enforces strict structured concurrency invariants using AnyIO 4 TaskGroups across FastAPI ASGI endpoints and service boundaries. Prevents orphaned background tasks, broken cancellation cascades, and uncontrolled exception fan-out across concurrent external I/O operations by mandating deterministic cancellation scopes and PEP 654 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: Unbounded asyncio.create_task and asyncio.gather Fan-Out Leaking Coroutines Across ASGI Worker Lifecycles 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.create_task and asyncio.gather Fan-Out Leaking Coroutines Across ASGI Worker Lifecycles	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 asynchronous I/O operations within an 'async with anyio.create_task_group() as tg

Engineers MUST ALWAYS encapsulate concurrent asynchronous I/O operations within an 'async with anyio.create_task_group() as tg:' context block.

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.