FastAPI Asyncio TaskGroup Structured Concurrency Guardrail

--- description: FastAPI Asyncio TaskGroup Structured Concurrency Guardrail - Enforces strict structured concurrency in FastAPI and Python 3.11+ services using asyncio.TaskGroup instead of fire-and-forget asyncio.create_task or unsupervised asyncio.gather. Guarantees that failure in any concurrent sub-task immediately triggers cooperative cancellation across sibling coroutines and propagates an ExceptionGroup, eliminating silent coroutine leaks, orphaned database connections, and zombie processing loops. globs: app/**/*.py, src/**/*.py, api/**/*.py, services/**/*.py alwaysApply: true verified_by: "agentrules.org" license: "CC-BY-4.0" source: "https://agentrules.org/rules/fastapi-asyncio-taskgroup-structured-concurrency" --- # FastAPI Asyncio TaskGroup Structured Concurrency Guardrail > **Severity**: CRITICAL | **Category**: backend > **Anti-Pattern Warning**: Unsupervised fire-and-forget asyncio.create_task or unconstrained asyncio.gather leading to zombie coroutines and swallowed exceptions ## System Constraints You are a Principal Python Architect specializing in asynchronous distributed systems and structured concurrency. Mandate the strict use of Python 3.11+ asyncio.TaskGroup for all fan-out/fan-in operations and asynchronous background job scheduling inside FastAPI services. Prohibit unhandled fire-and-forget coroutines (bare asyncio.create_task without retention or supervised lifecycle) and unsupervised asyncio.gather calls that suppress cancellation semantics or swallow critical downstream exceptions. Ensure all concurrent branches adhere to cooperative cancellation and cleanly unroll shared transactional contexts. ## Strict Guidelines - Engineers MUST ALWAYS manage concurrent fan-out tasks using 'async with asyncio.TaskGroup() as tg:' to guarantee bounded lifecycles and automatic sibling cancellation upon task failure. - Engineers MUST NOT invoke bare 'asyncio.create_task()' inside request lifecycles without either attaching the task to a supervised long-lived lifecycle registry or handling its completion inside a TaskGroup context. - Engineers MUST NOT use 'asyncio.gather(*tasks, return_exceptions=True)' as a blunt error-hiding mechanism; if errors are captured, engineers MUST inspect each result and explicitly handle or raise failures. - Services MUST catch multi-exception trees using Python 3.11 'except* ExceptionType' syntax when executing inside or exiting an asyncio.TaskGroup. - Components NEVER allow coroutines spawned inside request scope to outlive the ASGI HTTP response cycle unless explicitly queued into a durable background worker (e.g., Celery, ARQ, or a managed global background queue with graceful shutdown hooks). ## Bad vs Good Implementations ### Unstructured Fire-and-Forget Fan-Out vs. Supervised TaskGroup Concurrency #### Bad Antipattern (AI Common Hallucination) > **Why it fails**: Using bare asyncio.create_task and asyncio.gather with return_exceptions=True hides critical errors, breaks cooperative cancellation, and causes long-running sub-tasks to run as orphaned zombies when one fails, draining memory and connection pools. ```python import asyncio from fastapi import APIRouter, HTTPException router = APIRouter() async def fetch_enrichment(user_id: str) -> dict: # If this raises an unhandled exception, it logs a warning but leaves # audit logging hanging or leaking connections in the background await asyncio.sleep(0.1) raise RuntimeError("Enrichment service timed out") async def record_audit_metric(user_id: str) -> None: await asyncio.sleep(2.0) @router.post("/users/{user_id}/sync") async def sync_user(user_id: str): # ANTI-PATTERN: Bare create_task creates zombie coroutines. # Exception in fetch_enrichment is not surfaced to caller, audit task runs orphaned, # and exceptions can leak into the event loop root handler or be completely swallowed. t1 = asyncio.create_task(fetch_enrichment(user_id)) t2 = asyncio.create_task(record_audit_metric(user_id)) # Unsafe gather without cancellation guarantees results = await asyncio.gather(t1, t2, return_exceptions=True) return {"status": "completed", "results": str(results)} ``` #### Production Standard (Correct) > **Standard**: Leverages Python 3.11 asyncio.TaskGroup. When fetch_enrichment raises RuntimeError, record_audit_metric is cancelled immediately before TaskGroup exit. The resulting ExceptionGroup is handled cleanly with except* without leaking coroutines. ```python import asyncio import logging from fastapi import APIRouter, HTTPException, status logger = logging.getLogger(__name__) router = APIRouter() async def fetch_enrichment(user_id: str) -> dict: await asyncio.sleep(0.1) raise RuntimeError("Enrichment service timed out") async def record_audit_metric(user_id: str) -> None: # Cooperatively cancelled if sibling fails await asyncio.sleep(2.0) @router.post("/users/{user_id}/sync") async def sync_user(user_id: str): try: async with asyncio.TaskGroup() as tg: t_enrich = tg.create_task(fetch_enrichment(user_id)) t_audit = tg.create_task(record_audit_metric(user_id)) # Both tasks completed successfully if we reached this line enrichment_data = t_enrich.result() return {"status": "completed", "data": enrichment_data} except* RuntimeError as eg: logger.error("Enrichment task failed; siblings cooperatively cancelled: %s", eg.exceptions) raise HTTPException( status_code=status.HTTP_502_BAD_GATEWAY, detail="Dependent enrichment service failure" ) from eg ```

Architectural Specification & Invariant Rationale

Architectural Invariant Design: FastAPI Asyncio TaskGroup Structured Concurrency Guardrail

FastAPI Asyncio TaskGroup Structured Concurrency Guardrail defines an industrial-grade behavioral standard for backend engineering. Enforces strict structured concurrency in FastAPI and Python 3.11+ services using asyncio.TaskGroup instead of fire-and-forget asyncio.create_task or unsupervised asyncio.gather. Guarantees that failure in any concurrent sub-task immediately triggers cooperative cancellation across sibling coroutines and propagates an ExceptionGroup, eliminating silent coroutine leaks, orphaned database connections, and zombie processing 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: Unsupervised fire-and-forget asyncio.create_task or unconstrained asyncio.gather leading to zombie coroutines and swallowed exceptions 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	Unsupervised fire-and-forget asyncio.create_task or unconstrained asyncio.gather leading to zombie coroutines and swallowed exceptions	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 concurrent fan-out tasks using 'async with asyncio.TaskGroup() as tg

Engineers MUST ALWAYS manage concurrent fan-out tasks using 'async with asyncio.TaskGroup() as tg:' to guarantee bounded lifecycles and automatic sibling cancellation upon task failure.

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.