Async SQLAlchemy 2.0 Greenlet Context Leak & TaskGroup Isolation Guardrail

--- description: Async SQLAlchemy 2.0 Greenlet Context Leak & TaskGroup Isolation Guardrail - Enforces strict isolation of SQLAlchemy 2.0 AsyncSession instances across concurrent tasks inside Python 3.11+ asyncio.TaskGroup and AnyIO task groups within FastAPI. Prohibits sharing a single AsyncSession across child coroutines to prevent greenlet context corruption, cross-task connection state bleed, and 'concurrent operations are not permitted on a single Session' runtime exceptions. globs: **/*.py alwaysApply: true verified_by: "agentrules.org" license: "CC-BY-4.0" source: "https://agentrules.org/rules/sqlalchemy-async-taskgroup-greenlet-context-guardrail" --- # Async SQLAlchemy 2.0 Greenlet Context Leak & TaskGroup Isolation Guardrail > **Severity**: CRITICAL | **Category**: backend > **Anti-Pattern Warning**: Sharing a Request-Scoped AsyncSession across Concurrent TaskGroup Child Coroutines ## System Constraints You are a Principal Backend Architect enforcing strict concurrency rules for Async SQLAlchemy 2.0 and FastAPI. You must reject any implementation where a single `AsyncSession` is passed into or captured by concurrent coroutines inside `asyncio.TaskGroup` or AnyIO groups. Every parallel branch must instantiate its own isolated session spawned via an `async_sessionmaker`, and all parent models must be eagerly loaded before fan-out to prevent greenlet context corruption. ## Strict Guidelines - Engineers MUST NEVER pass a single `AsyncSession` instance into multiple concurrent tasks scheduled inside `asyncio.TaskGroup` or `anyio.create_task_group`. - Engineers MUST inject `async_sessionmaker[AsyncSession]` (not a live session) into background orchestrators or fan-out service layers to allow each parallel worker to spawn an independent session. - Child tasks executing database operations MUST manage their own isolated transaction lifecycle using `async with session_factory() as session: async with session.begin(): ...`. - Engines and session factories MUST configure `expire_on_commit=False` to prevent implicit greenlet-based lazy loading when referencing model attributes across coroutine context boundaries. - Queries returning ORM models intended for distribution across parallel tasks MUST explicitly execute with `selectinload` or be mapped to immutable Pydantic schemas/DTOs prior to entering the `TaskGroup`. ## Bad vs Good Implementations ### TaskGroup DB Fan-out: Shared Session vs. Per-Task Factory Sessions #### Bad Antipattern (AI Common Hallucination) > **Why it fails**: SQLAlchemy 2.0 AsyncSession wraps a single DBAPI connection via greenlet context. Sharing a single session across concurrent child tasks inside TaskGroup causes race conditions on the DBAPI connection, raising 'InvalidRequestError: Concurrent operations are not permitted on a single Session' and leaking the greenlet context across asynchronous request boundaries. ```python import asyncio from fastapi import APIRouter, Depends from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy import select from app.database import get_db from app.models import OrderItem router = APIRouter() async def process_item(session: AsyncSession, item_id: int) -> None: # CRITICAL BUG: Concurrent operations on the same session corrupt greenlet state stmt = select(OrderItem).where(OrderItem.id == item_id) result = await session.execute(stmt) item = result.scalar_one() item.is_processed = True await session.commit() @router.post("/orders/process-batch") async def process_batch( item_ids: list[int], session: AsyncSession = Depends(get_db) ) -> dict[str, str]: async with asyncio.TaskGroup() as tg: for item_id in item_ids: # Shared session passed to multiple concurrent tasks in taskgroup tg.create_task(process_item(session, item_id)) return {"status": "completed"} ``` #### Production Standard (Correct) > **Standard**: Injecting `async_sessionmaker` ensures every parallel task spawned by `TaskGroup` controls its own distinct `AsyncSession`, DB connection, and greenlet execution stack. Transactions are committed cleanly within the sub-task boundary without context pollution or race conditions. ```python import asyncio from fastapi import APIRouter, Depends from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker from sqlalchemy import select from app.database import get_session_factory from app.models import OrderItem router = APIRouter() async def process_item( session_factory: async_sessionmaker[AsyncSession], item_id: int ) -> None: # Isolated session and greenlet per child task branch async with session_factory() as session: async with session.begin(): stmt = select(OrderItem).where(OrderItem.id == item_id) result = await session.execute(stmt) item = result.scalar_one() item.is_processed = True # Transaction cleanly commits upon context manager exit @router.post("/orders/process-batch") async def process_batch( item_ids: list[int], session_factory: async_sessionmaker[AsyncSession] = Depends(get_session_factory) ) -> dict[str, str]: async with asyncio.TaskGroup() as tg: for item_id in item_ids: tg.create_task(process_item(session_factory, item_id)) return {"status": "completed"} ```

Architectural Specification & Invariant Rationale

Architectural Invariant Design: Async SQLAlchemy 2.0 Greenlet Context Leak & TaskGroup Isolation Guardrail

Async SQLAlchemy 2.0 Greenlet Context Leak & TaskGroup Isolation Guardrail defines an industrial-grade behavioral standard for backend engineering. Enforces strict isolation of SQLAlchemy 2.0 AsyncSession instances across concurrent tasks inside Python 3.11+ asyncio.TaskGroup and AnyIO task groups within FastAPI. Prohibits sharing a single AsyncSession across child coroutines to prevent greenlet context corruption, cross-task connection state bleed, and 'concurrent operations are not permitted on a single Session' runtime exceptions. 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: Sharing a Request-Scoped AsyncSession across Concurrent TaskGroup Child 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	Sharing a Request-Scoped AsyncSession across Concurrent TaskGroup Child 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 NEVER pass a single `AsyncSession` instance into multiple concurrent tasks scheduled inside `asyncio.TaskGroup` or `anyio.create_task_group`.

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.