FastAPI Pydantic v2 Async Generator Backpressure Standard

--- description: FastAPI Pydantic v2 Async Generator Backpressure Standard - Enforces strict TCP backpressure and bounded memory allocation for FastAPI StreamingResponse endpoints yielding Pydantic v2 serialized data. Eliminates fatal worker OOM events by coupling asynchronous database cursors and Pydantic serialization loops to ASGI send/drain cycles. globs: **/*.py alwaysApply: true verified_by: "agentrules.org" license: "CC-BY-4.0" source: "https://agentrules.org/rules/fastapi-pydantic-v2-async-generator-backpressure" --- # FastAPI Pydantic v2 Async Generator Backpressure Standard > **Severity**: CRITICAL | **Category**: backend > **Anti-Pattern Warning**: Unbounded Async Generator Generation Leading to Worker Heap Exhaustion via Unthrottled ASGI Response Buffering ## System Constraints You are a Principal Backend Architect enforcing strict backpressure and memory invariants in FastAPI and Pydantic v2 streaming systems. You must prevent unbounded heap bloat during multi-gigabyte exports by requiring explicit cursor pacing, bounded chunking, zero intermediate list aggregations, and synchronization with ASGI socket drain states. ## Strict Guidelines - Engineers MUST ALWAYS stream records using server-side cursors (e.g., asyncpg cursor, SQLAlchemy yield_per) with explicit batch fetch limits rather than loading entire datasets into memory. - Engineers MUST serialize Pydantic models directly to bytes or UTF-8 chunks using `model.model_dump_json().encode('utf-8')` on individual items or bounded chunks instead of serializing a root collection model. - Engineers MUST NOT yield unbounded raw generator streams without handling ASGI client disconnection and socket backpressure mechanisms. - Services MUST ensure that the consumer pull rate governs the producer fetch rate by awaiting ASGI `send` completion at each chunk yield point. - Code SHOULD enforce an explicit buffer size cap (e.g., high-water mark of 64KB-256KB) when aggregating line-delimited JSON (NDJSON) chunks to optimize packet density without ballooning heap usage. - Components NEVER instantiate intermediate in-memory lists, sets, or comprehension collections inside the streaming generator pipeline. ## Bad vs Good Implementations ### Unbounded Stream Generation vs Backpressure-Governed NDJSON Generator #### Bad Antipattern (AI Common Hallucination) > **Why it fails**: The generator produces JSON payloads at CPU speed without yielding execution to the underlying I/O loop or checking if the client has consumed previous packets. Uvicorn buffers ASGI message bodies in memory if the client socket is saturated or slow, rapidly blowing out the Python process heap into gigabytes. ```python from fastapi import FastAPI from fastapi.responses import StreamingResponse from pydantic import BaseModel from typing import AsyncGenerator class AuditLogSchema(BaseModel): id: int payload: str app = FastAPI() # ANTI-PATTERN: Eagerly serializing full models without chunking or backpressure # In starlette/uvicorn, fast generators queue thousands of chunks in ASGI output buffers # when clients read slowly (slow-loris / low bandwidth), exhausting worker RAM. async def unthrottled_producer(query_limit: int) -> AsyncGenerator[str, None]: # Simulates loading or rapidly yielding unbuffered stream for i in range(query_limit): record = AuditLogSchema(id=i, payload="x" * 1024) # Memory blowup: Starlette queues all yielded strings in memory yield record.model_dump_json() + "\n" @app.get("/export-bad") async def export_bad(): return StreamingResponse( unthrottled_producer(10_000_000), media_type="application/x-ndjson" ) ``` #### Production Standard (Correct) > **Standard**: Uses an internal bytearray accumulator with a defined high-water mark (64KB) to avoid tiny packet fragmentation. Checks `request.is_disconnected()` to preempt generation if the peer aborts, and executes `await asyncio.sleep(0)` after buffer yields, allowing the event loop to flush the TCP socket buffer and throttle generator progression to the client's consumption rate. ```python import asyncio from collections.abc import AsyncGenerator from fastapi import FastAPI, Request from fastapi.responses import StreamingResponse from pydantic import BaseModel class AuditLogSchema(BaseModel): id: int payload: str app = FastAPI() async def backpressure_bounded_generator( request: Request, total_records: int, chunk_byte_limit: int = 65536 # 64KB high-water mark ) -> AsyncGenerator[bytes, None]: buffer = bytearray() for i in range(total_records): # 1. Cooperative cancellation check on client disconnect if await request.is_disconnected(): break # 2. Serialize single instance to raw bytes via Pydantic v2 model = AuditLogSchema(id=i, payload="x" * 1024) chunk = model.model_dump_json().encode("utf-8") + b"\n" buffer.extend(chunk) # 3. Flushed only when watermark breached to preserve MTU efficiency if len(buffer) >= chunk_byte_limit: yield bytes(buffer) buffer.clear() # 4. Mandatory zero-sleep yield: lets asyncio event loop process TCP drain await asyncio.sleep(0) if buffer: yield bytes(buffer) @app.get("/export-good") async def export_good(request: Request): return StreamingResponse( backpressure_bounded_generator(request, total_records=10_000_000), media_type="application/x-ndjson", headers={"Content-Encoding": "identity", "X-Content-Type-Options": "nosniff"} ) ```

Architectural Specification & Invariant Rationale

Architectural Invariant Design: FastAPI Pydantic v2 Async Generator Backpressure Standard

FastAPI Pydantic v2 Async Generator Backpressure Standard defines an industrial-grade behavioral standard for backend engineering. Enforces strict TCP backpressure and bounded memory allocation for FastAPI StreamingResponse endpoints yielding Pydantic v2 serialized data. Eliminates fatal worker OOM events by coupling asynchronous database cursors and Pydantic serialization loops to ASGI send/drain cycles. 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: Unbounded Async Generator Generation Leading to Worker Heap Exhaustion via Unthrottled ASGI Response Buffering 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	Unbounded Async Generator Generation Leading to Worker Heap Exhaustion via Unthrottled ASGI Response Buffering	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 stream records using server-side cursors (e.g., asyncpg cursor, SQLAlchemy yield_per) with explicit batch fetch limits rather than loading entire datasets into memory.

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.