Agent-Skills.md

Agent Skills: Python Backend

Python backend patterns for asyncio, FastAPI, SQLAlchemy 2.0 async, and connection pooling. Use when building async Python services, FastAPI endpoints, database sessions, or connection pool tuning.

document-asset-creationID: yonatangross/orchestkit/python-backend

Repository

yonatangross/orchestkit
yonatangrossLicense: MIT
13314

Install this agent skill to your local

pnpm dlx add-skill https://github.com/yonatangross/orchestkit/tree/HEAD/src/skills/python-backend

Skill Files

Browse the full folder contents for python-backend.

Download Skill

Loading file tree…

src/skills/python-backend/SKILL.md

Skill Metadata

Name
python-backend
Description
"Production Python async patterns including asyncio TaskGroup, FastAPI dependency injection and middleware, SQLAlchemy 2.0 async sessions, and database connection pool tuning. Python 3.11+ examples with structured error handling. Use when building async services, FastAPI endpoints, or tuning database connection pools."
<!-- 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 | 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 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 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 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

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 separation
  • ork:async-jobs - Celery/ARQ for background processing
  • streaming-api-patterns - SSE/WebSocket async patterns
  • ork:database-patterns - Database schema design