/python-backend

name: python-backend
license: MIT
compatibility: "Claude Code 2.1.277+."
description: "Async Python: asyncio TaskGroup, FastAPI dependencies and middleware, SQLAlchemy 2.0 async sessions, pool tuning. Not for the API contract (use api-design)."
user-invocable: false
disable-model-invocation: false
targets:
  - library: fastapi
    version: ">=0.142.2"
  - library: sqlalchemy
    version: ">=2.1.1"
metadata:
  owner-agent: "backend-system-architect"
  category: document-asset-creation
  version: "2.0.0"
  author: "OrchestKit"
  complexity: "medium"
  tags: "python, asyncio, fastapi, sqlalchemy, connection-pooling, async, postgresql"
allowed-tools: "Read Glob Grep WebFetch WebSearch"
path_patterns: ["*.py", "**/requirements*.txt", "**/pyproject.toml", "**/Pipfile"]

<!-- directive-density: intentional (teaches asyncio/SQLAlchemy anti-patterns; NEVER markers describe real event-loop/race-condition bugs, not aspirational guidance) -->

Python Backend

Patterns for building production Python backends with asyncio, FastAPI, SQLAlchemy 2.0, and connection pooling. Each category has individual rule files in `rules/` loaded on-demand.

Quick Reference

| Category | Rules | Impact | When to Use | |----------|-------|--------|-------------| | [Asyncio](#asyncio) | 3 | HIGH | TaskGroup, structured concurrency, cancellation handling | | [FastAPI](#fastapi) | 3 | HIGH | Dependencies, middleware, background tasks | | [SQLAlchemy](#sqlalchemy) | 3 | HIGH | Async sessions, relationships, migrations | | [Pooling](#pooling) | 3 | MEDIUM | Database pools, HTTP sessions, tuning |

**Total: 12 rules across 4 categories.** House decisions rescued from thinned files live in `references/ork-delta.md`; vendor material is linked, not restated (see [Upstream coverage](#upstream-coverage-do-not-restate)).

Quick Start

# FastAPI + SQLAlchemy async session
async def get_db() -> AsyncGenerator[AsyncSession, None]:
    async with async_session_factory() as session:
        try:
            yield session
            await session.commit()
        except Exception:
            await session.rollback()
            raise

# Reusable dependency alias (FastAPI's recommended Annotated form)
SessionDep = Annotated[AsyncSession, Depends(get_db)]

@router.get("/users/{user_id}")
async def get_user(user_id: UUID, db: SessionDep):
    result = await db.execute(select(User).where(User.id == user_id))
    return result.scalar_one_or_none()

# Asyncio TaskGroup with timeout
async def fetch_all(urls: list[str]) -> list[dict]:
    async with asyncio.timeout(30):
        async with asyncio.TaskGroup() as tg:
            tasks = [tg.create_task(fetch_url(url)) for url in urls]
    return [t.result() for t in tasks]

Asyncio

Modern Python asyncio patterns using structured concurrency, TaskGroup, and Python 3.11+ features.

Key Patterns

• **TaskGroup** replaces `gather()` with structured concurrency and auto-cancellation
• **`asyncio.timeout()`** context manager for composable timeouts
• **Semaphore** for concurrency limiting (rate-limit HTTP requests)
• **`except*`** with ExceptionGroup for handling multiple task failures
• **`asyncio.to_thread()`** for bridging sync code to async

Key Decisions

| Decision | Recommendation | |----------|----------------| | Task spawning | TaskGroup not gather() | | Timeouts | asyncio.timeout() context manager | | Concurrency limit | asyncio.Semaphore | | Sync bridge | asyncio.to_thread() | | Cancellation | Always re-raise CancelledError |

FastAPI

Production-ready FastAPI patterns for lifespan, dependencies, middleware, and settings.

Key Patterns

• **Lifespan** with `asynccontextmanager` for startup/shutdown resource management
• **Dependency injection** with class-based services and `Depends()`
• **Middleware stack**: CORS -> RequestID -> Timing -> Logging
• **Pydantic Settings** with `.env` and field validation
• **Exception handlers** wired to RFC 9457 Problem Details bodies (the body format itself is `ork:api-design`)

Key Decisions

| Decision | Recommendation | |----------|----------------| | Lifespan | asynccontextmanager (not events) | | Dependencies | Class-based services with DI | | Settings | Pydantic Settings with .env | | Response | ORJSONResponse for performance | | Health | Check all critical dependencies |

SQLAlchemy

Async database patterns with SQLAlchemy 2.0, AsyncSession, and FastAPI integration.

Key Patterns

• **One AsyncSession per request** with `expire_on_commit=False`
• **`lazy="raise"`** on relationships to prevent accidental N+1 queries
• **`selectinload`** for eager loading collections
• **Repository pattern** with generic async CRUD
• **Bulk inserts** chunked 1000-10000 rows for memory management

Key Decisions

| Decision | Recommendation | |----------|----------------| | Session scope | One AsyncSession per request | | Lazy loading | lazy="raise" + explicit loads | | Eager loading | selectinload for collections | | expire_on_commit | False (prevents lazy load errors) | | Pool | pool_pre_ping=True |

Pooling

Database and HTTP connection pooling for high-performance async Python applications.

Key Patterns

• **SQLAlchemy pool** with `pool_size`, `max_overflow`, `pool_pre_ping`
• **Direct asyncpg pool** with `min_size`/`max_size` and connection lifecycle
• **aiohttp session** with `TCPConnector` limits and DNS caching
• **FastAPI lifespan** creating and closing pools at startup/shutdown
• **Pool monitoring** with Prometheus metrics

Pool Sizing Formula

pool_size = (concurrent_requests / avg_queries_per_request) * 1.5

That formula sizes one process. The fleet-level cap against the server's `max_connections`, and the pool alert thresholds, are in `references/ork-delta.md`.

Anti-Patterns (FORBIDDEN)

# NEVER use gather() for new code - no structured concurrency
# NEVER swallow CancelledError - breaks TaskGroup and timeout
# NEVER block the event loop with sync