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. House decisions rescued from thinned files live in references/ork-delta.md; vendor material is linked, not restated (see Upstream coverage).
Quick Start
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
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()
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
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 queriesselectinload 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)
Upstream coverage (do not restate)
Topics removed in the 2026-07-31 wrap-plus-delta thinning. Consult the first-party source; only the ork delta (house policy, scars, working config) belongs in this skill. Where a row says a house subset stays in a rules/ file, that file is still the authority for the OrchestKit position and the link only covers the vendor surface around it.
| Topic | First-party source |
|---|
asyncio task API reference: TaskGroup vs gather(), asyncio.timeout(), ExceptionGroup and except*, to_thread(). House subset stays in rules/asyncio-taskgroup.md and rules/asyncio-cancellation.md. | https://docs.python.org/3/library/asyncio-task.html |
asyncio.Semaphore, Lock, Event, Queue semantics. House subset stays in rules/asyncio-structured.md; the create-once-plus-timeout rule is in references/ork-delta.md. | https://docs.python.org/3/library/asyncio-sync.html |
Event-loop debug mode, slow_callback_duration, detecting blocking calls (the house 100 ms threshold is in references/ork-delta.md) | https://docs.python.org/3/library/asyncio-dev.html |
FastAPI project scaffolding: app layout, APIRouter composition, Pydantic Settings wiring, uvicorn entry point | https://fastapi.tiangolo.com/tutorial/bigger-applications/ |
FastAPI lifespan API and startup/shutdown mechanics. House subset stays in rules/fastapi-background.md; the reverse-order teardown rule is in references/ork-delta.md. | https://fastapi.tiangolo.com/advanced/events/ |
Starlette/FastAPI middleware API, BaseHTTPMiddleware, call_next, CORS options. House subset stays in rules/fastapi-middleware.md and references/fastapi-app-boilerplate.md; the middleware-vs-dependency split is in references/ork-delta.md. | https://fastapi.tiangolo.com/tutorial/middleware/ |
SQLAlchemy 2.0 asyncio API: create_async_engine, async_sessionmaker, expire_on_commit, /, , bulk insert and update. House subset stays in , , , and . |
Related Skills
ork:architecture-patterns - Clean architecture and layer separationork:async-jobs - Celery/ARQ for background processingork:api-design - Wire contract, RFC 9457 errors, SSE/WebSocket streamingork:database-patterns - Database schema designork:security-patterns - Auth, password hashing, token policyork:testing-integration - pytest-asyncio and httpx ASGI test setupork:devops-deployment - Docker and Kubernetes packaging
