FastAPI AnyIO Structured Concurrency and Lifespan Exception Shielding

--- description: FastAPI AnyIO Structured Concurrency and Lifespan Exception Shielding - Enforces strict structured concurrency in FastAPI application lifespans using AnyIO TaskGroups to prevent orphaned coroutines, unhandled task leakage, and silent process crashes. Guarantees clean resource draining and graceful shutdowns by wrapping resource teardown in cancel-shielded scopes during termination signals. globs: app/**/*.py, src/**/*.py, **/lifespan.py, **/main.py alwaysApply: true verified_by: "agentrules.org" license: "CC-BY-4.0" source: "https://agentrules.org/rules/fastapi-anyio-structured-concurrency-shielding" --- # FastAPI AnyIO Structured Concurrency and Lifespan Exception Shielding > **Severity**: CRITICAL | **Category**: backend > **Anti-Pattern Warning**: Spawning detached asyncio.create_task coroutines during lifespan and unshielded teardown execution ## System Constraints You are a Principal Backend Engineer enforcing AnyIO structured concurrency and cancel-safe lifespan resource management across FastAPI services. You must reject fire-and-forget task patterns such as raw asyncio.create_task in FastAPI lifespans, enforce cancel-shielded teardown routines, and mandate scoped TaskGroups. ## Strict Guidelines - Engineers MUST ALWAYS manage persistent background coroutines within an explicit AnyIO TaskGroup instantiated in the FastAPI lifespan. - Engineers MUST NOT invoke `asyncio.create_task` or `asyncio.ensure_future` for long-running workers during lifespan setup without deterministic supervisor bounds. - Engineers MUST wrap post-yield lifespan cleanup code in `with anyio.CancelScope(shield=True):` to guarantee critical teardown operations run uninterrupted during cancellation cascades. - Engineers MUST catch or properly propagate `ExceptionGroup` (or `BaseExceptionGroup`) when handling task failures spawned within the lifespan task group. - Background worker tasks SHOULD implement explicit cancellation checkpoints using `await anyio.sleep(0)` or `anyio.checkpoint()` inside tight execution loops. ## Bad vs Good Implementations ### Lifespan Background Coroutine Supervision & Shielded Teardown #### Bad Antipattern (AI Common Hallucination) > **Why it fails**: Spawning detached background tasks via asyncio.create_task decouples coroutines from the lifespan context. If metric_flusher raises an exception, it is unhandled until garbage collected. Furthermore, if teardown is interrupted by a cancellation signal, app.state.db.disconnect() is aborted midway, causing connection leaks. ```python import asyncio from contextlib import asynccontextmanager from fastapi import FastAPI async def metric_flusher(): while True: await asyncio.sleep(5) print("Flushing metrics...") @asynccontextmanager async def lifespan(app: FastAPI): # ANTI-PATTERN: Fire-and-forget detached coroutine. Leaks if app crashes or reloads; # exceptions in flusher vanish or crash unhandled. task = asyncio.create_task(metric_flusher()) yield # ANTI-PATTERN: Unshielded cancellation; if lifespan context is cancelled via SIGINT/SIGTERM, # database disconnect will be aborted halfway through. task.cancel() await app.state.db.disconnect() ``` #### Production Standard (Correct) > **Standard**: Using anyio.create_task_group() establishes strict structured concurrency. If metric_flusher fails, the exception group propagates deterministically. Wrapping the shutdown sequence in a shielded cancel scope guarantees database connections drain completely without being interrupted by cancellation signals. ```python from contextlib import asynccontextmanager from fastapi import FastAPI import anyio async def metric_flusher(): while True: await anyio.sleep(5) print("Flushing metrics...") @asynccontextmanager async def lifespan(app: FastAPI): async with anyio.create_task_group() as tg: # Scoped to lifespan: task lifecycle bound directly to the FastAPI server lifecycle tg.start_soon(metric_flusher) try: yield finally: # Shield cleanup to guarantee execution even during SIGTERM / cancellation cascades with anyio.CancelScope(shield=True): tg.cancel_scope.cancel() await app.state.db.disconnect() ```

Architectural Specification & Invariant Rationale

Architectural Invariant Design: FastAPI AnyIO Structured Concurrency and Lifespan Exception Shielding

FastAPI AnyIO Structured Concurrency and Lifespan Exception Shielding defines an industrial-grade behavioral standard for backend engineering. Enforces strict structured concurrency in FastAPI application lifespans using AnyIO TaskGroups to prevent orphaned coroutines, unhandled task leakage, and silent process crashes. Guarantees clean resource draining and graceful shutdowns by wrapping resource teardown in cancel-shielded scopes during termination signals. 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 detached asyncio.create_task coroutines during lifespan and unshielded teardown execution 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 detached asyncio.create_task coroutines during lifespan and unshielded teardown execution	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 manage persistent background coroutines within an explicit AnyIO TaskGroup instantiated in the FastAPI lifespan.

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.