FastAPI Async Native TaskGroup Concurrency Structural Isolation

--- description: FastAPI Async Native TaskGroup Concurrency Structural Isolation - Enforces structured concurrency invariants across FastAPI asynchronous request handlers using Python 3.12+ asyncio.TaskGroup. Mandates fail-fast cooperative cancellation, strict ExceptionGroup unwrapping, and deterministic cleanup to eliminate orphaned background coroutines and unbounded connection leaks during fan-out execution. globs: app/**/*.py, src/**/*.py, api/**/*.py alwaysApply: true verified_by: "agentrules.org" license: "CC-BY-4.0" source: "https://agentrules.org/rules/fastapi-taskgroup-structural-isolation-standard" --- # FastAPI Async Native TaskGroup Concurrency Structural Isolation > **Severity**: CRITICAL | **Category**: backend > **Anti-Pattern Warning**: Orphaned coroutine leakage and silent background execution via asyncio.gather with return_exceptions or unmanaged create_task fan-out ## System Constraints You are a Principal Software Architect enforcing Python 3.12+ asyncio.TaskGroup structured concurrency invariants within FastAPI services. Disallow legacy unstructured fan-out patterns such as asyncio.gather, bare asyncio.create_task inside request scopes without lifecycle boundaries, and unbounded concurrent execution. Mandate deterministic cancellation semantics, safe ExceptionGroup handling, resource scoping via async context managers, and strict isolation between independent fan-out branches. ## Strict Guidelines - Engineers MUST ALWAYS use 'async with asyncio.TaskGroup() as tg:' for dynamic parallel fan-out inside FastAPI route handlers and service layers. - Engineers MUST NOT rely on 'asyncio.gather(*tasks, return_exceptions=True)' for fan-out tasks holding system-critical resources (e.g., database transactions, network sockets) because it fails to cancel sibling coroutines when one fails. - Engineers MUST handle 'except* Exception' or specific sub-exception groups ('except* TimeoutError', 'except* httpx.HTTPError') to process structured multi-exception trees deterministically without discarding sibling context. - Services MUST bound concurrency within a TaskGroup using an 'asyncio.Semaphore' or worker pool pattern when dispatching high-cardinality tasks to prevent connection pool exhaustion. - Components NEVER spawn fire-and-forget tasks via 'asyncio.create_task' within an endpoint without anchoring them to an explicit lifecycle manager or background tasks queue. ## Bad vs Good Implementations ### Parallel Fan-Out in FastAPI Endpoint #### Bad Antipattern (AI Common Hallucination) > **Why it fails**: Using asyncio.create_task paired with asyncio.gather leaks background tasks on failure. When task_b throws an exception, gather aborts and re-raises immediately, but task_a and task_c remain scheduled as orphaned tasks on the event loop, leaking client sockets and processing cycles. ```python import asyncio import httpx from fastapi import APIRouter, HTTPException router = APIRouter() async def fetch_metrics(client: httpx.AsyncClient, source: str) -> dict: resp = await client.get(f"https://api.internal/{source}") resp.raise_for_status() return resp.json() @router.get("/aggregate") async def aggregate_data(): async with httpx.AsyncClient() as client: # ANTI-PATTERN: gather does not cancel sibling tasks upon failure # If 'source_b' raises, 'source_a' and 'source_c' continue executing blindly # and consume memory/network resources while the client already received an error. task_a = asyncio.create_task(fetch_metrics(client, "source_a")) task_b = asyncio.create_task(fetch_metrics(client, "source_b")) task_c = asyncio.create_task(fetch_metrics(client, "source_c")) try: results = await asyncio.gather(task_a, task_b, task_c) except Exception as exc: raise HTTPException(status_code=502, detail=str(exc)) return {"data": results} ``` #### Production Standard (Correct) > **Standard**: Using Python 3.12+ TaskGroup guarantees structured execution: if any task raises an exception, all remaining child tasks in the group are immediately and cooperatively cancelled. The except* clause cleanly unwraps the resulting ExceptionGroup. ```python import asyncio import httpx from fastapi import APIRouter, HTTPException, status router = APIRouter() async def fetch_metrics(client: httpx.AsyncClient, source: str, result_container: dict) -> None: resp = await client.get(f"https://api.internal/{source}", timeout=3.0) resp.raise_for_status() result_container[source] = resp.json() @router.get("/aggregate") async def aggregate_data(): results: dict[str, dict] = {} sources = ["source_a", "source_b", "source_c"] try: async with httpx.AsyncClient() as client: async with asyncio.TaskGroup() as tg: for source in sources: tg.create_task(fetch_metrics(client, source, results)) except* httpx.HTTPStatusError as eg: # Python 3.11+ ExceptionGroup pattern: sibling tasks were deterministically cancelled err_details = [f"{e.request.url.path}: {e.response.status_code}" for e in eg.exceptions] raise HTTPException( status_code=status.HTTP_502_BAD_GATEWAY, detail=f"Upstream failure in subtasks: {', '.join(err_details)}" ) except* httpx.RequestError as eg: raise HTTPException( status_code=status.HTTP_504_GATEWAY_TIMEOUT, detail=f"Network transport failure: {len(eg.exceptions)} tasks failed" ) return {"data": results} ```

Architectural Specification & Invariant Rationale

Architectural Invariant Design: FastAPI Async Native TaskGroup Concurrency Structural Isolation

FastAPI Async Native TaskGroup Concurrency Structural Isolation defines an industrial-grade behavioral standard for backend engineering. Enforces structured concurrency invariants across FastAPI asynchronous request handlers using Python 3.12+ asyncio.TaskGroup. Mandates fail-fast cooperative cancellation, strict ExceptionGroup unwrapping, and deterministic cleanup to eliminate orphaned background coroutines and unbounded connection leaks during fan-out execution. 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: Orphaned coroutine leakage and silent background execution via asyncio.gather with return_exceptions or unmanaged create_task fan-out 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	Orphaned coroutine leakage and silent background execution via asyncio.gather with return_exceptions or unmanaged create_task fan-out	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 'async with asyncio.TaskGroup() as tg

Engineers MUST ALWAYS use 'async with asyncio.TaskGroup() as tg:' for dynamic parallel fan-out inside FastAPI route handlers 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.