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 | 3 | HIGH | TaskGroup, structured concurrency, cancellation handling | | FastAPI | 3 | HIGH | Dependencies, middleware, background tasks | | SQLAlchemy | 3 | HIGH | Async sessions, relationships, migrations | | Pooling | 3 | MEDIUM | Database pools, HTTP sessions, tuning |
Total: 12 rules across 4 categories
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
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 failuresasyncio.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
asynccontextmanagerfor startup/shutdown resource management - Dependency injection with class-based services and
Depends() - Middleware stack: CORS -> RequestID -> Timing -> Logging
- Pydantic Settings with
.envand field validation - Exception handlers with RFC 9457 Problem Details
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 queriesselectinloadfor 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_sizeand connection lifecycle - aiohttp session with
TCPConnectorlimits 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
ork:architecture-patterns- Clean architecture and layer separationork:async-jobs- Celery/ARQ for background processingstreaming-api-patterns- SSE/WebSocket async patternsork:database-patterns- Database schema design