Python Backend

rules/

Quick Reference

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

@router.get("/users/{user_id}")
async def get_user(user_id: UUID, db: AsyncSession = Depends(get_db)):
    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

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

FastAPI

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 with RFC 7807 Problem Details

Key Decisions

SQLAlchemy

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

Pooling

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

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 calls (time.sleep, requests.get)
# NEVER use global mutable state for db sessions
# NEVER skip dependency injection (create sessions in routes)
# NEVER share AsyncSession across tasks (race condition)
# NEVER use sync Session in async code (blocks event loop)
# NEVER create engine/pool per request
# NEVER forget to close pools on shutdown

Related Skills

• architecture-patterns

  - Clean architecture and layer separation

• async-jobs

  - Celery/ARQ for background processing

• streaming-api-patterns

  - SSE/WebSocket async patterns

• database-patterns

  - Database schema design