FastAPI Async Generator Backpressure & Pydantic V2 Memory Invariant

--- description: FastAPI Async Generator Backpressure & Pydantic V2 Memory Invariant - Enforces strict backpressure handling and zero-accumulation memory bounds for Server-Sent Events (SSE) and chunked binary streams in FastAPI. Prevents unbounded heap growth and event loop starvation caused by unconstrained buffer accumulation, eager Pydantic V2 collection materialization, and missed client disconnection signals. globs: **/api/**/*.py, **/routers/**/*.py, **/services/**/*.py, **/streaming/**/*.py alwaysApply: true verified_by: "agentrules.org" license: "CC-BY-4.0" source: "https://agentrules.org/rules/fastapi-async-generator-backpressure-pydantic-v2-memory-invariant" --- # FastAPI Async Generator Backpressure & Pydantic V2 Memory Invariant > **Severity**: CRITICAL | **Category**: backend > **Anti-Pattern Warning**: Eager model collection materialization and unbounded queueing during SSE/chunked response streaming. ## System Constraints You are a Principal Backend Architect enforcing strict asyncio backpressure and zero-accumulation memory invariants for FastAPI endpoints utilizing Starlette StreamingResponse and Pydantic V2. You must reject any code pattern that materializes unbounded datasets into memory prior to yielding, consumes unbuffered or unbounded queues, omits client disconnect checks, or blocks the asyncio event loop during large serialization routines. ## Strict Guidelines - Engineers MUST ALWAYS check `await request.is_disconnected()` inside long-running or unbounded async generators to terminate production loops immediately upon client abort. - Engineers MUST NOT construct full lists of Pydantic models (e.g., `[MyModel.model_validate(row) for row in rows]`) prior to yielding chunks from an async generator. - Engineers MUST enforce bounded capacities on internal transport queues (e.g., `asyncio.Queue(maxsize=128)`) to propagate backpressure directly to upstream producers. - Code SHOULD stream directly via database cursor or async stream iterators (e.g., `AsyncResult.stream()` or `aiter()`), validating and serializing single items on-the-fly using `model.model_dump_json()`. - Services MUST NOT run synchronous CPU-bound transformations or large JSON encodings inside the streaming loop without yielding control to the event loop using `await asyncio.sleep(0)` or delegating to worker threads. - Components NEVER buffer streaming responses into intermediate `list` or `bytes` aggregations within the application heap before dispatching to `StreamingResponse`. ## Bad vs Good Implementations ### Unbounded List Materialization vs. Lazy Chunked Backpressure Stream #### Bad Antipattern (AI Common Hallucination) > **Why it fails**: Fetches the complete dataset into memory at once and validates all Pydantic models up front, exhausting RAM. Does not check if the client disconnected, leading to wasted processing, and provides zero backpressure to the database. ```python from fastapi import APIRouter, Request from fastapi.responses import StreamingResponse from pydantic import BaseModel from typing import AsyncGenerator class TelemetryEvent(BaseModel): device_id: str payload: dict router = APIRouter() @router.get("/events/stream") async def bad_stream_events(request: Request): # CRITICAL FLAW: Eagerly fetches all records and instantiates entire collection in memory. # No backpressure, no disconnect check, unbounded heap allocation causing OOM. raw_records = await db.fetch_all_records() events = [TelemetryEvent.model_validate(r) for r in raw_records] async def event_generator() -> AsyncGenerator[str, None]: for event in events: yield f"data: {event.model_dump_json()}\n\n" return StreamingResponse(event_generator(), media_type="text/event-stream") ``` #### Production Standard (Correct) > **Standard**: Processes rows lazily one-by-one via database cursor, validates models on-demand, checks for client disconnection on each iteration, yields control to prevent event loop starvation, and disables reverse proxy response buffering via `X-Accel-Buffering: no`. ```python from fastapi import APIRouter, Request from fastapi.responses import StreamingResponse from pydantic import BaseModel import asyncio from typing import AsyncGenerator class TelemetryEvent(BaseModel): device_id: str payload: dict router = APIRouter() @router.get("/events/stream") async def good_stream_events(request: Request) -> StreamingResponse: async def event_generator() -> AsyncGenerator[str, None]: # Invariant: Lazy cursor iteration prevents unbounded heap allocation async for row in db.stream_cursor(): # Invariant: Terminate generator immediately if client disconnects if await request.is_disconnected(): break # Invariant: Single-instance validation and direct serialization event = TelemetryEvent.model_validate(row) yield f"data: {event.model_dump_json()}\n\n" # Invariant: Yield control back to event loop to allow socket flushing await asyncio.sleep(0) return StreamingResponse( event_generator(), media_type="text/event-stream", headers={ "Cache-Control": "no-cache", "Connection": "keep-alive", "X-Accel-Buffering": "no" } ) ```

Architectural Specification & Invariant Rationale

Architectural Invariant Design: FastAPI Async Generator Backpressure & Pydantic V2 Memory Invariant

FastAPI Async Generator Backpressure & Pydantic V2 Memory Invariant defines an industrial-grade behavioral standard for backend engineering. Enforces strict backpressure handling and zero-accumulation memory bounds for Server-Sent Events (SSE) and chunked binary streams in FastAPI. Prevents unbounded heap growth and event loop starvation caused by unconstrained buffer accumulation, eager Pydantic V2 collection materialization, and missed client disconnection signals. 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: Eager model collection materialization and unbounded queueing during SSE/chunked response streaming. 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	Eager model collection materialization and unbounded queueing during SSE/chunked response streaming.	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 check `await request.is_disconnected()` inside long-running or unbounded async generators to terminate production loops immediately upon client abort.

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.