FastAPI AnyIO Structured Concurrency & TaskGroup Invariant Standard

--- description: FastAPI AnyIO Structured Concurrency & TaskGroup Invariant Standard - Enforces strict structured concurrency in FastAPI route handlers and domain services via AnyIO TaskGroups and deterministic cancellation scopes. Prohibits fire-and-forget background asyncio tasks that cause silent task abandonment, orphan database sessions, and connection pool exhaustion. 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-anyio-structured-concurrency-taskgroup-standard" --- # FastAPI AnyIO Structured Concurrency & TaskGroup Invariant Standard > **Severity**: CRITICAL | **Category**: backend > **Anti-Pattern Warning**: Orphan Asyncio Tasks Leaking Database Connection Pools and Silently Swallowing Failures ## System Constraints You are a Principal Backend Architect enforcing strict structured concurrency standards across FastAPI applications powered by AnyIO. You must eliminate all instances of unstructured concurrency—including asyncio.create_task, asyncio.gather without return_exceptions management, and improper FastAPI BackgroundTasks sharing active transactional database sessions. You mandate that any parallel or asynchronous child task must be managed within an explicit anyio.create_task_group() context with bounded cancellation scopes and strictly separated unit-of-work sessions. ## Strict Guidelines - Engineers MUST NOT spawn detached fire-and-forget background tasks using 'asyncio.create_task()' or 'asyncio.ensure_future()' within route handlers or services. - Engineers MUST wrap all concurrent child operations within an 'async with anyio.create_task_group() as tg:' context to guarantee deterministic task completion or coordinated cancellation before request exit. - Engineers MUST NEVER pass a request-scoped database session (e.g., SQLAlchemy 'AsyncSession') to background child tasks; child tasks MUST resolve their own independent session factories or units of work. - Code SHOULD wrap non-critical parallel tasks with 'anyio.move_on_after(timeout)' or explicit 'tg.cancel_scope' boundaries to prevent cascading request timeouts from freezing connection pools. - Services MUST re-raise or handle 'anyio.ExceptionGroup' (or BaseExceptionGroup in Python 3.11+) when handling collective errors from task groups, ensuring clean rollback of distributed resources. - Endpoints MUST use persistent task queue brokers (e.g., Celery, ARQ, Temporal) for work intended to outlive the HTTP response lifecycle instead of unbounded in-process concurrency. ## Bad vs Good Implementations ### Fire-and-Forget asyncio.create_task vs AnyIO TaskGroup with Independent Sessions #### Bad Antipattern (AI Common Hallucination) > **Why it fails**: asyncio.create_task launches unmonitored tasks. The shared AsyncSession is accessed concurrently, violating SQLAlchemy thread/task safety, and child tasks outlive client disconnection, stranding pooled DB connections. ```python import asyncio from fastapi import APIRouter, Depends from sqlalchemy.ext.asyncio import AsyncSession from app.database import get_db from app.services import audit_service, notification_service router = APIRouter() @router.post("/orders") async def create_order(payload: dict, db: AsyncSession = Depends(get_db)): # Anti-Pattern: Fire-and-forget task referencing the request-scoped DB session # Causes 'InterfaceError: cannot perform operation: another operation is in progress' # and leaks connection pool slots when client aborts asyncio.create_task(audit_service.log_order(db, payload)) asyncio.create_task(notification_service.dispatch(payload)) return {"status": "accepted"} ``` #### Production Standard (Correct) > **Standard**: AnyIO TaskGroup binds child tasks to the current execution scope. Each task requiring database persistence consumes an independent session from async_session_factory, avoiding concurrent session misuse and eliminating leaks. ```python import anyio from fastapi import APIRouter, Depends, status from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker from app.database import get_db, async_session_factory from app.services import audit_service, notification_service router = APIRouter() async def _isolated_audit(session_maker: async_sessionmaker, payload: dict) -> None: async with session_maker() as session: async with session.begin(): await audit_service.log_order(session, payload) @router.post("/orders", status_code=status.HTTP_202_ACCEPTED) async def create_order(payload: dict, db: AsyncSession = Depends(get_db)): # Correct: AnyIO TaskGroup guarantees child tasks terminate or cancel deterministically # before the response completes. Independent session isolation prevents race conditions. async with anyio.create_task_group() as tg: tg.start_soon(_isolated_audit, async_session_factory, payload) tg.start_soon(notification_service.dispatch, payload) return {"status": "processed"} ```

Architectural Specification & Invariant Rationale

Architectural Invariant Design: FastAPI AnyIO Structured Concurrency & TaskGroup Invariant Standard

FastAPI AnyIO Structured Concurrency & TaskGroup Invariant Standard defines an industrial-grade behavioral standard for backend engineering. Enforces strict structured concurrency in FastAPI route handlers and domain services via AnyIO TaskGroups and deterministic cancellation scopes. Prohibits fire-and-forget background asyncio tasks that cause silent task abandonment, orphan database sessions, and connection pool exhaustion. 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: Orphan Asyncio Tasks Leaking Database Connection Pools and Silently Swallowing Failures 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	Orphan Asyncio Tasks Leaking Database Connection Pools and Silently Swallowing Failures	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 NOT spawn detached fire-and-forget background tasks using 'asyncio.create_task()' or 'asyncio.ensure_future()' within route handlers or services.

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.