FastAPI Async TaskGroup Structured Concurrency & Exception Isolation Boundary

--- description: FastAPI Async TaskGroup Structured Concurrency & Exception Isolation Boundary - Enforces strict structured concurrency in FastAPI asynchronous operations using Python 3.11+ asyncio.TaskGroup instead of raw asyncio.create_task or fire-and-forget background execution. Ensures that any unhandled exception immediately cancels sibling child tasks, isolates failures to the request lifecycle boundary, and prevents event loop corruption or orphaned coroutines leaking memory and resources. 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-structured-concurrency-boundary" --- # FastAPI Async TaskGroup Structured Concurrency & Exception Isolation Boundary > **Severity**: CRITICAL | **Category**: backend > **Anti-Pattern Warning**: Unbounded fire-and-forget asyncio.create_task leaking orphaned coroutines and suppressing unhandled ExceptionGroups. ## System Constraints You are a Principal Backend Architect enforcing structured concurrency and exception isolation invariants in FastAPI services. Forbid raw asyncio.create_task and asyncio.gather without strict cancellation semantics. Enforce Python 3.11+ asyncio.TaskGroup with explicit ExceptionGroup unwrapping or Starlette BackgroundTasks for fire-and-forget telemetry. ## Strict Guidelines - Engineers MUST ALWAYS encapsulate concurrent child coroutines inside an 'async with asyncio.TaskGroup() as tg:' context block. - Engineers MUST NOT spawn detached coroutines using 'asyncio.create_task()' inside request handlers without retaining explicit task references and attaching cancellation hooks. - Services MUST catch and inspect 'ExceptionGroup' (or BaseExceptionGroup) explicitly when handling TaskGroup failures to log all underlying errors individually. - Engineers MUST NOT use 'asyncio.gather(..., return_exceptions=True)' as a substitute for structured concurrency when failure in any single dependency invalidates the entire composite operation. - Components NEVER leak background workloads beyond the HTTP request-response boundary unless explicitly managed through Starlette BackgroundTasks or an external durable queue (e.g., Celery, ARQ, Temporal). ## Bad vs Good Implementations ### FastAPI Concurrent Fan-Out: Raw asyncio.create_task vs. asyncio.TaskGroup Boundary #### Bad Antipattern (AI Common Hallucination) > **Why it fails**: If t2 fails while t1 is awaiting, an unhandled exception will raise prematurely or orphan t1, leaving it running in the event loop. Furthermore, unhandled exceptions in raw tasks can escape without proper cancellation, exhausting connection pools and causing silent state leakage. ```python import asyncio from fastapi import APIRouter, HTTPException router = APIRouter() async def fetch_user_data(user_id: str) -> dict: await asyncio.sleep(0.05) return {"id": user_id} async def fetch_billing_data(user_id: str) -> dict: await asyncio.sleep(0.02) raise ConnectionResetError("Billing service unreachable") @router.get("/users/{user_id}/dashboard") async def get_dashboard(user_id: str): # ANTI-PATTERN: Detached create_task calls orphan sibling tasks on failure, # leak memory, and suppress exceptions until garbage collection. t1 = asyncio.create_task(fetch_user_data(user_id)) t2 = asyncio.create_task(fetch_billing_data(user_id)) # If fetch_billing_data fails, fetch_user_data continues running unmonitored user_res = await t1 billing_res = await t2 return {"user": user_res, "billing": billing_res} ``` #### Production Standard (Correct) > **Standard**: Using Python 3.11's asyncio.TaskGroup establishes a deterministic scope boundary. If fetch_billing_data raises ConnectionResetError, fetch_user_data is immediately cancelled, preventing orphaned coroutines and connection pool leakage. The except* syntax cleanly unwraps the resulting ExceptionGroup. ```python import asyncio import logging from fastapi import APIRouter, HTTPException, status logger = logging.getLogger(__name__) router = APIRouter() async def fetch_user_data(user_id: str) -> dict: await asyncio.sleep(0.05) return {"id": user_id} async def fetch_billing_data(user_id: str) -> dict: await asyncio.sleep(0.02) raise ConnectionResetError("Billing service unreachable") @router.get("/users/{user_id}/dashboard") async def get_dashboard(user_id: str): user_res: dict | None = None billing_res: dict | None = None async def _safe_user(): nonlocal user_res user_res = await fetch_user_data(user_id) async def _safe_billing(): nonlocal billing_res billing_res = await fetch_billing_data(user_id) try: # INVARIANT: TaskGroup guarantees fail-fast cancellation of all child tasks # and deterministic resource cleanup at scope exit. async with asyncio.TaskGroup() as tg: tg.create_task(_safe_user()) tg.create_task(_safe_billing()) except* ConnectionResetError as eg: logger.error("Upstream dependency failed during dashboard fan-out: %s", eg.exceptions) raise HTTPException( status_code=status.HTTP_503_SERVICE_UNAVAILABLE, detail="Billing service temporarily unavailable" ) except* Exception as eg: logger.critical("Unexpected concurrent failure: %s", eg.exceptions) raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="Internal server error during data aggregation" ) return {"user": user_res, "billing": billing_res} ```

Architectural Specification & Invariant Rationale

Architectural Invariant Design: FastAPI Async TaskGroup Structured Concurrency & Exception Isolation Boundary

FastAPI Async TaskGroup Structured Concurrency & Exception Isolation Boundary defines an industrial-grade behavioral standard for backend engineering. Enforces strict structured concurrency in FastAPI asynchronous operations using Python 3.11+ asyncio.TaskGroup instead of raw asyncio.create_task or fire-and-forget background execution. Ensures that any unhandled exception immediately cancels sibling child tasks, isolates failures to the request lifecycle boundary, and prevents event loop corruption or orphaned coroutines leaking memory and resources. 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 fire-and-forget asyncio.create_task leaking orphaned coroutines and suppressing unhandled ExceptionGroups. 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 fire-and-forget asyncio.create_task leaking orphaned coroutines and suppressing unhandled ExceptionGroups.	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 inside an 'async with asyncio.TaskGroup() as tg

Engineers MUST ALWAYS encapsulate concurrent child coroutines inside an 'async with asyncio.TaskGroup() 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.