FastAPI AnyIO Structured TaskGroup Exception Shielding and Context Preservation Invariants

--- description: FastAPI AnyIO Structured TaskGroup Exception Shielding and Context Preservation Invariants - Enforces robust structured concurrency patterns within FastAPI asynchronous request lifecycles using AnyIO TaskGroups. Prevents unhandled child-task exceptions and cancellation cascades from aborting critical audit, telemetry, and transactional cleanup routines or propagating raw ExceptionGroups into ASGI worker loops. globs: app/**/*.py, src/**/*.py, services/**/*.py, api/**/*.py alwaysApply: true verified_by: "agentrules.org" license: "CC-BY-4.0" source: "https://agentrules.org/rules/fastapi-anyio-taskgroup-exception-shielding" --- # FastAPI AnyIO Structured TaskGroup Exception Shielding and Context Preservation Invariants > **Severity**: CRITICAL | **Category**: backend > **Anti-Pattern Warning**: Spawning naked concurrent child tasks within unshielded AnyIO TaskGroups, causing unhandled ExceptionGroups to tear down sibling operations and drop audit context. ## System Constraints You are a Principal Software Architect enforcing production-grade concurrency invariants for FastAPI applications utilizing AnyIO. You MUST strictly prohibit naked, unhandled background tasks and unshielded task groups in request lifecycles. Mandate structured exception handling using ExceptionGroup/except* syntax, explicit cancellation shields for audit and persistent state commits, and strict contextvar propagation. ## Strict Guidelines - Engineers MUST ALWAYS wrap AnyIO TaskGroup blocks with granular exception handlers that properly unpack ExceptionGroup or BaseExceptionGroup to prevent ASGI worker crashes. - Engineers MUST ALWAYS enclose mission-critical audit logging, billing finalization, and security trace persistence within an `anyio.CancelScope(shield=True)` context to prevent cancellation cascade truncation. - Engineers MUST NOT spawn unbounded concurrent tasks inside request scopes using raw `asyncio.create_task` or naked `tg.start_soon` without an internal per-task error boundary wrapper. - Services MUST ensure that Python `contextvars` (e.g., correlation ID, tenant ID, authenticated user token) are explicitly retained and isolated across task boundaries. - Components NEVER allow a non-critical child task failure (such as an analytics or notification hook) to bubble up and abort a parent financial or database transaction. ## Bad vs Good Implementations ### Unshielded AnyIO TaskGroup vs. Shielded Error-Bound Concurrency #### Bad Antipattern (AI Common Hallucination) > **Why it fails**: If `emit_audit_log` raises an exception, the TaskGroup immediately triggers a cancellation cascade on sibling tasks (canceling `process_payment` midway through execution). The resulting unhandled ExceptionGroup bubbles up to the ASGI server, dropping execution context and generating an unhandled 500 error without finalized audit records. ```python import anyio from fastapi import APIRouter, HTTPException router = APIRouter() async def emit_audit_log(order_id: str): # If this fails, it collapses the entire task group await anyio.sleep(0.01) raise RuntimeError("Audit service down") async def process_payment(order_id: str): await anyio.sleep(0.05) return {"status": "paid"} @router.post("/orders/{order_id}/checkout") async def checkout(order_id: str): # ANTI-PATTERN: Unshielded TaskGroup causes emit_audit_log failure # to cancel process_payment midway, bubbling an raw ExceptionGroup to Uvicorn async with anyio.create_task_group() as tg: tg.start_soon(process_payment, order_id) tg.start_soon(emit_audit_log, order_id) return {"message": "Order processed"} ``` #### Production Standard (Correct) > **Standard**: The safe variant employs an internal cancellation shield (`CancelScope(shield=True)`) and an explicit error boundary inside the audit sub-task. If any unhandled exception does escape, Python 3.11+ `except*` structured exception matching catches the `ExceptionGroup`, records detailed diagnostics, and formats a clean ASGI response rather than crashing the worker event loop. ```python import logging import anyio from fastapi import APIRouter, HTTPException, status logger = logging.getLogger(__name__) router = APIRouter() async def safe_emit_audit_log(order_id: str) -> None: # Invariant: Non-critical operations MUST have internal error boundaries # and run in a shielded scope if invoked during cleanup/teardown. with anyio.CancelScope(shield=True): try: await anyio.sleep(0.01) # Simulated external service call except Exception as exc: logger.error("Audit emission failed for order %s: %s", order_id, exc, exc_info=True) async def execute_payment(order_id: str) -> dict: await anyio.sleep(0.05) return {"status": "paid"} @router.post("/orders/{order_id}/checkout") async def checkout(order_id: str): payment_result: dict = {} try: async with anyio.create_task_group() as tg: # Shielded critical audit task will not drop even if caller is cancelled tg.start_soon(safe_emit_audit_log, order_id) payment_result = await execute_payment(order_id) except* Exception as eg: # Invariant: Structured exception unpacking prevents worker loop teardown logger.critical("Unhandled concurrency fault in checkout: %s", eg.exceptions) raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="Checkout processing encountered an unrecoverable failure." ) from eg return {"message": "Order processed", "payment": payment_result} ```

Architectural Specification & Invariant Rationale

Architectural Invariant Design: FastAPI AnyIO Structured TaskGroup Exception Shielding and Context Preservation Invariants

FastAPI AnyIO Structured TaskGroup Exception Shielding and Context Preservation Invariants defines an industrial-grade behavioral standard for backend engineering. Enforces robust structured concurrency patterns within FastAPI asynchronous request lifecycles using AnyIO TaskGroups. Prevents unhandled child-task exceptions and cancellation cascades from aborting critical audit, telemetry, and transactional cleanup routines or propagating raw ExceptionGroups into ASGI worker loops. 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 naked concurrent child tasks within unshielded AnyIO TaskGroups, causing unhandled ExceptionGroups to tear down sibling operations and drop audit context. 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 naked concurrent child tasks within unshielded AnyIO TaskGroups, causing unhandled ExceptionGroups to tear down sibling operations and drop audit context.	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 wrap AnyIO TaskGroup blocks with granular exception handlers that properly unpack ExceptionGroup or BaseExceptionGroup to prevent ASGI worker crashes.

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.