FastAPI Structured Concurrency TaskGroup Exception Boundary Standard

--- description: FastAPI Structured Concurrency TaskGroup Exception Boundary Standard - Enforces structured concurrency boundaries in FastAPI applications using Python 3.11+ asyncio.TaskGroup. Prevents orphaned coroutines, silent task failures, and database connection pool starvation caused by fire-and-forget tasks or improperly shared AsyncSession lifecycles. globs: app/**/*.py, src/**/*.py, api/**/*.py alwaysApply: true verified_by: "agentrules.org" license: "CC-BY-4.0" source: "https://agentrules.org/rules/fastapi-taskgroup-concurrency-boundary-standard" --- # FastAPI Structured Concurrency TaskGroup Exception Boundary Standard > **Severity**: CRITICAL | **Category**: backend > **Anti-Pattern Warning**: Fire-and-forget asyncio.create_task and passing active AsyncSession instances to unbounded background coroutines. ## System Constraints You are a Principal Backend Architect enforcing strict Structured Concurrency invariants in FastAPI. Mandate asyncio.TaskGroup for intra-request concurrent tasks, isolate AsyncSession lifecycles per task, and eliminate unmonitored asyncio.create_task invocations to prevent connection pool exhaustion and orphaned processing loops. ## Strict Guidelines - Engineers MUST ALWAYS use 'asyncio.TaskGroup' or scoped worker primitives instead of unshielded 'asyncio.create_task' for intra-request sub-operations. - Engineers MUST NOT pass a request-scoped database session (e.g., SQLAlchemy 'AsyncSession') into background tasks or concurrent branches. - Concurrent child tasks MUST manage their own independent database session context managers created from an 'async_sessionmaker'. - Services MUST catch and unwrap 'ExceptionGroup' or 'BaseExceptionGroup' using 'except* ' syntax (Python 3.11+) or explicitly map child failures before completing request boundaries. - FastAPI endpoints MUST NOT initiate unbounded async workflows that outlive the HTTP response lifecycle; external task queues (e.g., Celery, ARQ, SAQ) MUST be used for detached background processing. ## Bad vs Good Implementations ### Unbounded fire-and-forget task leaking AsyncSession vs Structured TaskGroup with scoped session #### Bad Antipattern (AI Common Hallucination) > **Why it fails**: asyncio.create_task runs without a boundary. If an exception occurs, it is silenced until garbage collection. Furthermore, borrowing the request-scoped AsyncSession causes race conditions, premature connection closure, or pool starvation. ```python import asyncio from fastapi import APIRouter, Depends from sqlalchemy.ext.asyncio import AsyncSession from app.db import get_db router = APIRouter() async def log_audit_trail(session: AsyncSession, user_id: str): # Reuses caller session; will fail or corrupt connection state when request scope closes await session.execute("INSERT INTO audit_logs (user_id) VALUES (:id)", {"id": user_id}) await session.commit() @router.post("/users/{user_id}/action") async def perform_action(user_id: str, db: AsyncSession = Depends(get_db)): # ANTI-PATTERN: Fire-and-forget create_task with borrowed session asyncio.create_task(log_audit_trail(db, user_id)) return {"status": "accepted"} ``` #### Production Standard (Correct) > **Standard**: Using asyncio.TaskGroup enforces an execution barrier ensuring child tasks terminate before the parent scope exits. The child task provisions its own AsyncSession instance, preventing cross-coroutine socket contention and connection pool leaks. ```python import asyncio from fastapi import APIRouter, Depends, HTTPException, status from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker from app.db import get_db, sessionmaker router = APIRouter() async def record_audit_isolated(session_factory: async_sessionmaker[AsyncSession], user_id: str) -> None: async with session_factory() as session: async with session.begin(): await session.execute("INSERT INTO audit_logs (user_id) VALUES (:id)", {"id": user_id}) @router.post("/users/{user_id}/action") async def perform_action(user_id: str, db: AsyncSession = Depends(get_db)): try: # STRUCTURED CONCURRENCY: TaskGroup manages child cancellation and exception boundaries async with asyncio.TaskGroup() as tg: tg.create_task(record_audit_isolated(sessionmaker, user_id)) except* Exception as eg: # Python 3.11+ exception group handling prevents silent task drops raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"Concurrent operations failed: {eg.exceptions}" ) return {"status": "completed"} ```

Architectural Specification & Invariant Rationale

Architectural Invariant Design: FastAPI Structured Concurrency TaskGroup Exception Boundary Standard

FastAPI Structured Concurrency TaskGroup Exception Boundary Standard defines an industrial-grade behavioral standard for backend engineering. Enforces structured concurrency boundaries in FastAPI applications using Python 3.11+ asyncio.TaskGroup. Prevents orphaned coroutines, silent task failures, and database connection pool starvation caused by fire-and-forget tasks or improperly shared AsyncSession lifecycles. 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: Fire-and-forget asyncio.create_task and passing active AsyncSession instances to unbounded background coroutines. 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	Fire-and-forget asyncio.create_task and passing active AsyncSession instances to unbounded background coroutines.	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 'asyncio.TaskGroup' or scoped worker primitives instead of unshielded 'asyncio.create_task' for intra-request sub-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.