FastAPI Lifespan Structured Concurrency with Python 3.12+ TaskGroup

--- description: FastAPI Lifespan Structured Concurrency with Python 3.12+ TaskGroup - Enforces strict structured concurrency within FastAPI application lifespans using Python 3.12+ asyncio.TaskGroup and ASGI state propagation. Eliminates leaked zombie coroutines, dropped telemetry/event batches, and non-deterministic SIGTERM teardowns during container orchestration draining. globs: **/main.py, **/lifespan.py, **/app/**/*.py, **/core/**/*.py alwaysApply: true verified_by: "agentrules.org" license: "CC-BY-4.0" source: "https://agentrules.org/rules/fastapi-lifespan-taskgroup-structured-concurrency" --- # FastAPI Lifespan Structured Concurrency with Python 3.12+ TaskGroup > **Severity**: CRITICAL | **Category**: backend > **Anti-Pattern Warning**: Unbounded asyncio.create_task in FastAPI routes/events leading to orphaned coroutines on SIGTERM shutdown ## System Constraints You are a Principal Backend Systems Architect enforcing absolute structured concurrency across FastAPI and ASGI lifespans. Never permit naked asyncio.create_task calls without TaskGroup ownership or lifespan lifecycle coordination. All background daemons and recurring event loops must be bound to a lifespan-managed asyncio.TaskGroup with coordinated termination flags to ensure safe SIGTERM process drainage. ## Strict Guidelines - Engineers MUST ALWAYS use Python 3.12+ async with asyncio.TaskGroup() inside an @asynccontextmanager lifespan handler to spawn and manage application-level background coroutines. - Engineers MUST NOT use deprecated @app.on_event('startup') or @app.on_event('shutdown') handlers. - Engineers MUST NOT spawn detached background processing via bare asyncio.create_task() without binding task lifecycles to an application-managed TaskGroup or explicit task registry. - Lifespan managers MUST yield state via a typed dictionary or Mapping, propagating an asyncio.Event shutdown barrier to signal graceful task draining upon receiving SIGTERM/SIGINT. - Worker loops inside background tasks MUST check the shutdown asyncio.Event on every iteration and complete outstanding in-flight work before terminating. - Lifespan exception handling MUST explicitly account for BaseExceptionGroup / ExceptionGroup unwrapping to prevent silent container crashes during teardown. ## Bad vs Good Implementations ### FastAPI Lifespan Structured Concurrency with TaskGroup vs Fire-and-Forget create_task #### Bad Antipattern (AI Common Hallucination) > **Why it fails**: Using bare asyncio.create_task() inside lifespan or route handlers causes coroutines to run unanchored. When the container receives a SIGTERM, Uvicorn drains HTTP connections and shuts down the loop, abruptly severing in-flight flushes and dropping critical logs, traces, or database commits without notification. ```python import asyncio from contextlib import asynccontextmanager from fastapi import FastAPI # ANTI-PATTERN: Fire-and-forget task without lifecycle ownership or shutdown hooks @asynccontextmanager async def lifespan(app: FastAPI): # Leaked task: on SIGTERM, Uvicorn terminates the event loop, # killing this coroutine mid-execution without flushing buffers. asyncio.create_task(flush_metrics_daemon()) yield async def flush_metrics_daemon(): while True: await asyncio.sleep(5) print("Flushing metrics...") app = FastAPI(lifespan=lifespan) ``` #### Production Standard (Correct) > **Standard**: Utilizes Python 3.12+ asyncio.TaskGroup wrapped around the lifespan yield. The background daemon explicitly observes an asyncio.Event shutdown flag. Upon ASGI shutdown (SIGTERM), the finally block signals the event, the loop exits gracefully, and TaskGroup guarantees all tasks finish draining before the application process exits. ```python import asyncio from contextlib import asynccontextmanager from typing import AsyncIterator, TypedDict from fastapi import FastAPI, Request class LifespanState(TypedDict): shutdown_event: asyncio.Event async def flush_metrics_daemon(shutdown_event: asyncio.Event) -> None: """Background worker strictly bound to the lifespan termination signal.""" while not shutdown_event.is_set(): try: # Bounded sleep that immediately wakes up on shutdown trigger await asyncio.wait_for(shutdown_event.wait(), timeout=5.0) except* TimeoutError: # Regular polling tick await asyncio.sleep(0.01) # Simulate non-blocking atomic flush print("Flushing metrics atomically...") # Final draining sequence prior to task exit print("Lifespan shutdown received: successfully drained remaining telemetry.") @asynccontextmanager async def lifespan(app: FastAPI) -> AsyncIterator[LifespanState]: shutdown_event = asyncio.Event() state: LifespanState = {"shutdown_event": shutdown_event} try: # Python 3.12+ TaskGroup guarantees deterministic cleanup async with asyncio.TaskGroup() as tg: tg.create_task( flush_metrics_daemon(shutdown_event), name="metrics-daemon-worker" ) try: yield state finally: # Signal all cooperative child workers to drain and stop shutdown_event.set() except* Exception as eg: # Handle ExceptionGroup if child workers raised during shutdown print(f"Encountered exceptions during lifespan drain: {eg.exceptions}") raise app = FastAPI(lifespan=lifespan) @app.get("/healthz") async def health_check(request: Request) -> dict[str, str]: is_draining = request.state.shutdown_event.is_set() return {"status": "terminating" if is_draining else "healthy"} ```

Architectural Specification & Invariant Rationale

Architectural Invariant Design: FastAPI Lifespan Structured Concurrency with Python 3.12+ TaskGroup

FastAPI Lifespan Structured Concurrency with Python 3.12+ TaskGroup defines an industrial-grade behavioral standard for backend engineering. Enforces strict structured concurrency within FastAPI application lifespans using Python 3.12+ asyncio.TaskGroup and ASGI state propagation. Eliminates leaked zombie coroutines, dropped telemetry/event batches, and non-deterministic SIGTERM teardowns during container orchestration draining. 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 in FastAPI routes/events leading to orphaned coroutines on SIGTERM shutdown 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 in FastAPI routes/events leading to orphaned coroutines on SIGTERM shutdown	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 Python 3.12+ async with asyncio.TaskGroup() inside an @asynccontextmanager lifespan handler to spawn and manage application-level background coroutines.

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.