Agent Skills: Python Backend

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

# 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 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

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, Mapped/mapped_column, with_for_update, bulk insert and update. House subset stays in rules/sqlalchemy-sessions.md, rules/sqlalchemy-relationships.md, rules/sqlalchemy-migrations.md, and references/eager-loading.md. | https://docs.sqlalchemy.org/en/20/orm/extensions/asyncio.html | | SQLAlchemy pool implementations and options (pool_size, max_overflow, pool_pre_ping, pool_recycle, pool_timeout). House subset stays in rules/pooling-database.md. | https://docs.sqlalchemy.org/en/20/core/pooling.html | | SQLAlchemy pool events (checkout, checkin, connect) for instrumentation. House subset stays in rules/pooling-tuning.md; the alert thresholds are in references/ork-delta.md. | https://docs.sqlalchemy.org/en/20/core/events.html | | asyncpg pool API: create_pool, min_size/max_size, max_inactive_connection_lifetime, setup hook, type codecs | https://magicstack.github.io/asyncpg/current/api/index.html | | aiohttp client reference: ClientSession, TCPConnector limits, keep-alive, DNS caching, ClientTimeout. House subset stays in rules/pooling-http.md. | https://docs.aiohttp.org/en/stable/client_reference.html | | PostgreSQL server-side connection limits (max_connections, superuser_reserved_connections) | https://www.postgresql.org/docs/current/runtime-config-connection.html | | pytest-asyncio fixtures and async test setup | https://pytest-asyncio.readthedocs.io/en/stable/reference/fixtures/index.html | | Container and Kubernetes packaging, readiness/liveness probes, resource limits | https://kubernetes.io/docs/tasks/configure-pod-container/configure-liveness-readiness-startup-probes/ | | Auth flows, password hashing, token expiry policy, security headers | Owned by ork:security-patterns (src/skills/security-patterns/); the FastAPI Depends() auth chain stays in rules/fastapi-dependencies.md | | RFC 9457 problem-details body format, API versioning, SSE and WebSocket wire contracts | Owned by ork:api-design (src/skills/api-design/); the FastAPI exception-handler wiring stays in rules/fastapi-background.md and references/fastapi-app-boilerplate.md | | Production checklists for FastAPI, asyncio, SQLAlchemy, and pooling | Derivable from the sources above; no checklist restatement kept |

Related Skills

• ork:architecture-patterns - Clean architecture and layer separation
• ork:async-jobs - Celery/ARQ for background processing
• ork:api-design - Wire contract, RFC 9457 errors, SSE/WebSocket streaming
• ork:database-patterns - Database schema design
• ork:security-patterns - Auth, password hashing, token policy
• ork:testing-integration - pytest-asyncio and httpx ASGI test setup
• ork:devops-deployment - Docker and Kubernetes packaging