FastAPI AnyIO Structured Concurrency and Task-Group Boundary Enforcement

--- description: FastAPI AnyIO Structured Concurrency and Task-Group Boundary Enforcement - Enforces strict AnyIO task group boundaries across all parallel async I/O worker pipelines in FastAPI applications. Mandates deterministic task cancellation, prevents unmonitored background tasks that leak connection pool slots, and requires bidirectional exception-group unwrapping using Python 3.11+ ExceptionGroup or anyio.create_task_group. globs: app/**/*.py, src/**/*.py, api/**/*.py alwaysApply: true verified_by: "agentrules.org" license: "CC-BY-4.0" source: "https://agentrules.org/rules/fastapi-anyio-structured-concurrency-boundary" --- # FastAPI AnyIO Structured Concurrency and Task-Group Boundary Enforcement > **Severity**: CRITICAL | **Category**: backend > **Anti-Pattern Warning**: Unstructured asyncio.create_task / asyncio.gather in FastAPI causing orphaned coroutines and connection pool starvation on client disconnects or worker exceptions ## System Constraints You are a Principal Backend Systems Architect specializing in Python high-throughput microservices. Mandate strict structured concurrency invariants using AnyIO task groups within FastAPI handlers and background pipelines. Enforce deterministic task cancellation cascades, eliminate asyncio.create_task() or raw asyncio.gather() calls that escape request-lifecycle lifetimes, and mandate structured unwrapping of ExceptionGroup/BaseExceptionGroup to guarantee DB pool connection liberation. ## Strict Guidelines - Engineers MUST ALWAYS use `async with anyio.create_task_group() as tg:` to coordinate multiple concurrent I/O operations. - Engineers MUST NOT spawn unbounded concurrent workers via `asyncio.create_task` or bare `asyncio.gather(..., return_exceptions=True)` inside FastAPI endpoints or service dependencies. - Engineers MUST handle `ExceptionGroup` (or `anyio.ExceptionGroup`) explicitly using `except*` (Python 3.11+) or `tg.cancel_scope.cancel()` to propagate deterministic cancellation signals down worker trees. - Services MUST bind database session, HTTP client session, or messaging channels within context-managed boundaries that guarantee cleanup even when cancelled via `anyio.get_cancelled_exc_class()`. - Components NEVER swallow or suppress `BaseExceptionGroup` or `anyio.get_cancelled_exc_class()` within broad `except Exception:` catches, which inhibits graceful task cancellation. ## Bad vs Good Implementations ### Unstructured gather vs AnyIO TaskGroup with pool safety and deterministic cancellation #### Bad Antipattern (AI Common Hallucination) > **Why it fails**: asyncio.create_task combined with asyncio.gather without structured boundaries risks leaving dangling coroutines when client disconnect occurs or when one item raises an exception. Unhandled exceptions in worker tasks are not propagated immediately to siblings, causing database connection leaks and pool starvation. ```python import asyncio from fastapi import APIRouter, Depends from sqlalchemy.ext.asyncio import AsyncSession from app.dependencies import get_db, get_http_client router = APIRouter() @router.post("/batch-sync") async def sync_data(items: list[str], db: AsyncSession = Depends(get_db)): # ANTI-PATTERN: Fire-and-forget or unstructured asyncio.gather leaks DB/HTTP sessions # If one task fails, remaining tasks continue running orphaned in the background # Swallows client disconnect cancellations, leading to connection pool exhaustion async def process_item(item: str): async with get_http_client() as client: resp = await client.post("https://external.service/api", json={"id": item}) await db.execute("INSERT INTO logs (data) VALUES (:d)", {"d": resp.text}) await db.commit() # Unsafe concurrent execution without lifecycle boundaries tasks = [asyncio.create_task(process_item(item)) for item in items] results = await asyncio.gather(*tasks, return_exceptions=True) return {"synced": len(results)} ``` #### Production Standard (Correct) > **Standard**: Using anyio.create_task_group ensures that if any single worker fails, all sibling tasks receive an immediate cancellation signal via AnyIO's cancel scope. Database sessions are managed per unit-of-work inside deterministic scopes, eliminating pool exhaustion on partial failures. ```python import anyio from fastapi import APIRouter, Depends, HTTPException, status from sqlalchemy.ext.asyncio import AsyncSession from app.dependencies import get_db_factory, get_http_client router = APIRouter() @router.post("/batch-sync") async def sync_data(items: list[str], db_factory = Depends(get_db_factory)): # INVARIANT: AnyIO TaskGroup establishes an airtight cancellation boundary # If any child task fails, the task group automatically cancels all peer tasks async def process_item(item: str, task_group: anyio.abc.TaskGroup): async with db_factory() as session: async with get_http_client() as client: resp = await client.post("https://external.service/api", json={"id": item}) resp.raise_for_status() await session.execute("INSERT INTO logs (data) VALUES (:d)", {"d": resp.text}) await session.commit() try: async with anyio.create_task_group() as tg: for item in items: tg.start_soon(process_item, item, tg) except* Exception as eg: # Structured unwrapping ensures all DB sessions and sockets were closed on cancel raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"Parallel batch processing failed: {len(eg.exceptions)} sub-task errors" ) return {"status": "completed", "count": len(items)} ```

Architectural Specification & Invariant Rationale

Architectural Invariant Design: FastAPI AnyIO Structured Concurrency and Task-Group Boundary Enforcement

FastAPI AnyIO Structured Concurrency and Task-Group Boundary Enforcement defines an industrial-grade behavioral standard for backend engineering. Enforces strict AnyIO task group boundaries across all parallel async I/O worker pipelines in FastAPI applications. Mandates deterministic task cancellation, prevents unmonitored background tasks that leak connection pool slots, and requires bidirectional exception-group unwrapping using Python 3.11+ ExceptionGroup or anyio.create_task_group. 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: Unstructured asyncio.create_task / asyncio.gather in FastAPI causing orphaned coroutines and connection pool starvation on client disconnects or worker 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	Unstructured asyncio.create_task / asyncio.gather in FastAPI causing orphaned coroutines and connection pool starvation on client disconnects or worker 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 use `async with anyio.create_task_group() as tg

Engineers MUST ALWAYS use `async with anyio.create_task_group() as tg:` to coordinate multiple concurrent I/O operations.

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.