Python Project

Python Project Architecture

Core Principles

• Type hints everywhere — Pydantic for runtime, mypy for static
• uv for everything — Package management, virtualenv, Python version
• Ruff only — Replace Flake8 + Black + isort with single tool
• src layout — All code under src/ directory
• pyproject.toml only — No setup.py, no requirements.txt
• Async all the way — Once async, stay async through call chain
• No backwards compatibility — Delete, don't deprecate. Change directly
• LiteLLM for LLM APIs — Use LiteLLM proxy for all LLM integrations

---

No Backwards Compatibility

Delete unused code. Change directly. No compatibility layers.

# ❌ BAD: Deprecated decorator kept around
import warnings

def old_function():
    warnings.warn("Use new_function instead", DeprecationWarning)
    return new_function()

# ❌ BAD: Alias for renamed functions
new_name = old_name  # "for backwards compatibility"

# ❌ BAD: Unused parameters with underscore
def process(_legacy_param, data):
    ...

# ❌ BAD: Version checking for old behavior
if version < "2.0":
    # old behavior
    ...

# ✅ GOOD: Just delete and update all usages
def new_function():
    ...
# Then: Find & replace all old_function → new_function

# ✅ GOOD: Remove unused parameters entirely
def process(data):
    ...

---

LiteLLM for LLM APIs

Use LiteLLM proxy. Don't call provider APIs directly.

# src/myapp/llm.py
from openai import AsyncOpenAI

from myapp.config import settings

# Connect to LiteLLM proxy using OpenAI SDK
client = AsyncOpenAI(
    base_url=settings.litellm_url,  # "http://localhost:4000"
    api_key=settings.litellm_api_key,
)


async def complete(prompt: str, model: str = "gpt-4o") -> str:
    """Call any LLM through LiteLLM proxy."""
    response = await client.chat.completions.create(
        model=model,  # "gpt-4o", "claude-3-opus", "gemini-pro", etc.
        messages=[{"role": "user", "content": prompt}],
    )
    return response.choices[0].message.content or ""

---

Quick Start

1. Initialize Project

# Install uv (if not installed)
curl -LsSf https://astral.sh/uv/install.sh | sh

# Create new project
uv init myapp
cd myapp

# Set Python version
echo "3.12" > .python-version

# Add dependencies
uv add fastapi uvicorn pydantic sqlalchemy httpx
uv add --dev pytest pytest-asyncio ruff mypy

2. Apply Tech Stack

Version Strategy

Always use latest. Never pin in templates.

[project]
dependencies = [
    "fastapi",      # uv resolves to latest
    "pydantic",
    "sqlalchemy",
]

• uv add fetches latest compatible versions
• uv.lock ensures reproducible builds
• uv sync installs exact locked versions

3. Use Standard Structure (src layout)

myapp/
├── pyproject.toml         # Single config file
├── uv.lock                # Lock file (commit this)
├── .python-version        # Python version for uv
├── src/
│   └── myapp/
│       ├── __init__.py
│       ├── __main__.py    # Entry point
│       ├── main.py        # FastAPI app
│       ├── config.py      # Pydantic Settings
│       ├── models/        # Pydantic models
│       │   ├── __init__.py
│       │   └── user.py
│       ├── services/      # Business logic
│       │   ├── __init__.py
│       │   └── user.py
│       ├── repositories/  # Data access
│       │   ├── __init__.py
│       │   └── user.py
│       ├── api/           # HTTP layer
│       │   ├── __init__.py
│       │   ├── deps.py    # Dependencies
│       │   └── routes/
│       │       ├── __init__.py
│       │       └── user.py
│       └── core/          # Shared utilities
│           ├── __init__.py
│           ├── exceptions.py
│           └── logging.py
├── tests/
│   ├── __init__.py
│   ├── conftest.py        # Fixtures
│   └── test_user.py
└── Makefile

---

Architecture Layers

main.py — FastAPI Application

# src/myapp/main.py
from contextlib import asynccontextmanager

from fastapi import FastAPI

from myapp.api.routes import router
from myapp.config import settings
from myapp.core.logging import setup_logging
from myapp.db import engine


@asynccontextmanager
async def lifespan(app: FastAPI):
    # Startup
    setup_logging()
    yield
    # Shutdown
    await engine.dispose()


app = FastAPI(
    title=settings.app_name,
    lifespan=lifespan,
)

app.include_router(router, prefix="/api/v1")


@app.get("/health")
async def health():
    return {"status": "ok"}

config.py — Pydantic Settings

# src/myapp/config.py
from pydantic_settings import BaseSettings, SettingsConfigDict


class Settings(BaseSettings):
    model_config = SettingsConfigDict(
        env_file=".env",
        env_file_encoding="utf-8",
    )

    app_name: str = "myapp"
    debug: bool = False

    # Database
    database_url: str = "postgresql+asyncpg://localhost/myapp"

    # LiteLLM
    litellm_url: str = "http://localhost:4000"
    litellm_api_key: str = ""


settings = Settings()

models/ — Pydantic Models

# src/myapp/models/user.py
from datetime import datetime
from uuid import UUID

from pydantic import BaseModel, EmailStr, Field


class UserBase(BaseModel):
    email: EmailStr
    name: str = Field(min_length=2, max_length=100)


class UserCreate(UserBase):
    pass


class UserUpdate(BaseModel):
    email: EmailStr | None = None
    name: str | None = Field(default=None, min_length=2, max_length=100)


class User(UserBase):
    id: UUID
    created_at: datetime
    updated_at: datetime

    model_config = {"from_attributes": True}

services/ — Business Logic

# src/myapp/services/user.py
from uuid import UUID

from myapp.core.exceptions import NotFoundError, ConflictError
from myapp.models.user import User, UserCreate, UserUpdate
from myapp.repositories.user import UserRepository


class UserService:
    def __init__(self, repo: UserRepository):
        self.repo = repo

    async def get(self, id: UUID) -> User:
        user = await self.repo.get(id)
        if not user:
            raise NotFoundError("user", str(id))
        return user

    async def create(self, data: UserCreate) -> User:
        existing = await self.repo.get_by_email(data.email)
        if existing:
            raise ConflictError("email already exists")
        return await self.repo.create(data)

    async def update(self, id: UUID, data: UserUpdate) -> User:
        user = await self.get(id)
        return await self.repo.update(user, data)

    async def delete(self, id: UUID) -> None:
        user = await self.get(id)
        await self.repo.delete(user)

api/routes/ — HTTP Handlers

# src/myapp/api/routes/user.py
from uuid import UUID

from fastapi import APIRouter, Depends, status

from myapp.api.deps import get_user_service
from myapp.models.user import User, UserCreate, UserUpdate
from myapp.services.user import UserService

router = APIRouter(prefix="/users", tags=["users"])


@router.get("/{id}", response_model=User)
async def get_user(
    id: UUID,
    service: UserService = Depends(get_user_service),
):
    return await service.get(id)


@router.post("", response_model=User, status_code=status.HTTP_201_CREATED)
async def create_user(
    data: UserCreate,
    service: UserService = Depends(get_user_service),
):
    return await service.create(data)


@router.patch("/{id}", response_model=User)
async def update_user(
    id: UUID,
    data: UserUpdate,
    service: UserService = Depends(get_user_service),
):
    return await service.update(id, data)


@router.delete("/{id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_user(
    id: UUID,
    service: UserService = Depends(get_user_service),
):
    await service.delete(id)

core/exceptions.py — Custom Exceptions

# src/myapp/core/exceptions.py
from fastapi import HTTPException, status


class AppError(Exception):
    """Base application error."""

    def __init__(self, message: str, code: str):
        self.message = message
        self.code = code
        super().__init__(message)


class NotFoundError(AppError):
    def __init__(self, resource: str, id: str):
        super().__init__(f"{resource} not found: {id}", "NOT_FOUND")


class ConflictError(AppError):
    def __init__(self, message: str):
        super().__init__(message, "CONFLICT")


class ValidationError(AppError):
    def __init__(self, message: str):
        super().__init__(message, "VALIDATION_ERROR")


# FastAPI exception handler
def app_error_to_http(error: AppError) -> HTTPException:
    status_map = {
        "NOT_FOUND": status.HTTP_404_NOT_FOUND,
        "CONFLICT": status.HTTP_409_CONFLICT,
        "VALIDATION_ERROR": status.HTTP_400_BAD_REQUEST,
    }
    return HTTPException(
        status_code=status_map.get(error.code, status.HTTP_500_INTERNAL_SERVER_ERROR),
        detail={"message": error.message, "code": error.code},
    )

---

pyproject.toml

[project]
name = "myapp"
version = "0.1.0"
description = "My application"
requires-python = ">=3.12"
dependencies = [
    "fastapi",
    "uvicorn[standard]",
    "pydantic",
    "pydantic-settings",
    "sqlalchemy[asyncio]",
    "asyncpg",
    "httpx",
    "structlog",
]

[tool.uv]
dev-dependencies = [
    "pytest",
    "pytest-asyncio",
    "pytest-cov",
    "ruff",
    "mypy",
]

[tool.ruff]
line-length = 100
target-version = "py312"

[tool.ruff.lint]
select = [
    "E",   # pycodestyle errors
    "F",   # pyflakes
    "I",   # isort
    "UP",  # pyupgrade
    "B",   # flake8-bugbear
    "SIM", # flake8-simplify
]

[tool.ruff.lint.isort]
known-first-party = ["myapp"]

[tool.mypy]
strict = true
python_version = "3.12"

[tool.pytest.ini_options]
asyncio_mode = "auto"
testpaths = ["tests"]

---

Extended Reference

Detailed material starting at ## Testing has been moved to `reference/extended.md` to keep this skill concise. Load that reference when the task requires the moved examples, command catalogs, checklists, platform details, or implementation templates.

Python Project Architecture

Project Layouts

src Layout (Recommended)

The src/ layout prevents import confusion and ensures you're testing installed code.

myapp/
├── pyproject.toml
├── src/
│   └── myapp/           # Package lives here
│       ├── __init__.py
│       └── ...
├── tests/
│   └── ...
└── docs/

Why src layout?

• Prevents accidental imports from project root
• Forces pip install -e . for development
• Clean separation between package and project files
• Wheel/sdist only includes src/ contents

Flat Layout (Simple Scripts)

For quick scripts or small tools:

myapp/
├── pyproject.toml
├── myapp/               # Package at root
│   ├── __init__.py
│   └── main.py
└── tests/

---

Application Structure

Web API (FastAPI)

src/myapp/
├── __init__.py
├── __main__.py          # python -m myapp
├── main.py              # FastAPI app factory
├── config.py            # Pydantic Settings
│
├── api/                 # HTTP layer
│   ├── __init__.py
│   ├── deps.py          # Dependency injection
│   ├── middleware.py    # Custom middleware
│   └── routes/
│       ├── __init__.py  # Router aggregation
│       ├── user.py
│       └── health.py
│
├── models/              # Pydantic schemas
│   ├── __init__.py
│   ├── base.py          # Base models
│   ├── user.py
│   └── common.py        # Shared types
│
├── services/            # Business logic
│   ├── __init__.py
│   └── user.py
│
├── repositories/        # Data access
│   ├── __init__.py
│   ├── base.py          # Repository interface
│   └── user.py
│
├── db/                  # Database
│   ├── __init__.py
│   ├── session.py       # Engine & session
│   ├── models.py        # SQLAlchemy models
│   └── migrations/      # Alembic
│
└── core/                # Shared utilities
    ├── __init__.py
    ├── exceptions.py
    ├── logging.py
    └── security.py

CLI Application

src/mycli/
├── __init__.py
├── __main__.py          # Entry point
├── cli.py               # Click/Typer commands
├── config.py            # Configuration
│
├── commands/            # Subcommands
│   ├── __init__.py
│   ├── init.py
│   └── process.py
│
├── services/            # Business logic
│   └── ...
│
└── core/                # Shared utilities
    ├── exceptions.py
    └── logging.py

Data Pipeline

src/pipeline/
├── __init__.py
├── main.py              # Pipeline entry
├── config.py
│
├── extractors/          # Data sources
│   ├── __init__.py
│   ├── base.py
│   └── api.py
│
├── transformers/        # Data processing
│   ├── __init__.py
│   └── clean.py
│
├── loaders/             # Data destinations
│   ├── __init__.py
│   └── db.py
│
└── models/              # Data models
    └── ...

---

Dependency Flow

┌─────────────────────────────────────────────┐
│                    API                       │
│              (routes, deps)                  │
└─────────────────────────────────────────────┘
                     │
                     ▼
┌─────────────────────────────────────────────┐
│                 Services                     │
│            (business logic)                  │
└─────────────────────────────────────────────┘
                     │
          ┌──────────┼──────────┐
          ▼          ▼          ▼
    ┌──────────┐ ┌──────────┐ ┌──────────┐
    │  Repos   │ │ External │ │  Cache   │
    │          │ │   APIs   │ │          │
    └──────────┘ └──────────┘ └──────────┘
          │
          ▼
    ┌──────────┐
    │    DB    │
    └──────────┘

Rules:

• API layer only handles HTTP concerns
• Services contain all business logic
• Repositories abstract data access
• Lower layers never import from upper layers

---

Module Organization

__init__.py Exports

# src/myapp/models/__init__.py
from myapp.models.user import User, UserCreate, UserUpdate
from myapp.models.common import Pagination

__all__ = [
    "User",
    "UserCreate",
    "UserUpdate",
    "Pagination",
]

Route Aggregation

# src/myapp/api/routes/__init__.py
from fastapi import APIRouter

from myapp.api.routes import health, user

router = APIRouter()
router.include_router(health.router)
router.include_router(user.router, prefix="/users", tags=["users"])

Entry Points

# src/myapp/__main__.py
"""Allow running as: python -m myapp"""
import uvicorn

from myapp.config import settings


def main():
    uvicorn.run(
        "myapp.main:app",
        host="0.0.0.0",
        port=settings.port,
        reload=settings.debug,
    )


if __name__ == "__main__":
    main()

---

Configuration

Pydantic Settings

# src/myapp/config.py
from functools import lru_cache

from pydantic import Field, PostgresDsn
from pydantic_settings import BaseSettings, SettingsConfigDict


class Settings(BaseSettings):
    model_config = SettingsConfigDict(
        env_file=".env",
        env_file_encoding="utf-8",
        case_sensitive=False,
    )

    # App
    app_name: str = "myapp"
    debug: bool = False
    port: int = 8000

    # Database
    database_url: PostgresDsn = Field(
        default="postgresql+asyncpg://localhost/myapp"
    )

    # External services
    litellm_url: str = "http://localhost:4000"
    litellm_api_key: str = ""

    # Secrets (from env only, not .env file)
    secret_key: str = Field(default="changeme")


@lru_cache
def get_settings() -> Settings:
    return Settings()


settings = get_settings()

Environment Files

# .env.example (commit this)
APP_NAME=myapp
DEBUG=false
PORT=8000
DATABASE_URL=postgresql+asyncpg://user:pass@localhost/myapp
LITELLM_URL=http://localhost:4000
LITELLM_API_KEY=

# .env (never commit)
DATABASE_URL=postgresql+asyncpg://prod:secret@db.example.com/myapp
SECRET_KEY=super-secret-key

---

Database Layer

SQLAlchemy 2.0 Async

# src/myapp/db/session.py
from sqlalchemy.ext.asyncio import (
    AsyncSession,
    async_sessionmaker,
    create_async_engine,
)

from myapp.config import settings

engine = create_async_engine(
    str(settings.database_url),
    echo=settings.debug,
)

async_session = async_sessionmaker(
    engine,
    class_=AsyncSession,
    expire_on_commit=False,
)


async def get_session() -> AsyncSession:
    async with async_session() as session:
        yield session

SQLAlchemy Models

# src/myapp/db/models.py
from datetime import datetime
from uuid import UUID, uuid4

from sqlalchemy import String, func
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column


class Base(DeclarativeBase):
    pass


class UserModel(Base):
    __tablename__ = "users"

    id: Mapped[UUID] = mapped_column(primary_key=True, default=uuid4)
    email: Mapped[str] = mapped_column(String(255), unique=True, index=True)
    name: Mapped[str] = mapped_column(String(100))
    created_at: Mapped[datetime] = mapped_column(server_default=func.now())
    updated_at: Mapped[datetime] = mapped_column(
        server_default=func.now(),
        onupdate=func.now(),
    )

---

Testing Structure

tests/
├── __init__.py
├── conftest.py          # Shared fixtures
├── factories.py         # Test data factories
│
├── unit/                # Unit tests (isolated)
│   ├── __init__.py
│   ├── test_services.py
│   └── test_models.py
│
├── integration/         # Integration tests (with DB)
│   ├── __init__.py
│   ├── conftest.py      # DB fixtures
│   └── test_repositories.py
│
└── e2e/                 # End-to-end tests (full API)
    ├── __init__.py
    └── test_api.py

conftest.py

# tests/conftest.py
import pytest
from httpx import ASGITransport, AsyncClient

from myapp.main import app


@pytest.fixture
async def client():
    """Async test client for API tests."""
    async with AsyncClient(
        transport=ASGITransport(app=app),
        base_url="http://test",
    ) as client:
        yield client


@pytest.fixture
def user_data():
    """Sample user data for tests."""
    return {
        "email": "test@example.com",
        "name": "Test User",
    }

---

Monorepo / Multi-Package

For larger projects with multiple packages:

myproject/
├── pyproject.toml       # Workspace config
├── uv.lock
│
├── packages/
│   ├── core/            # Shared library
│   │   ├── pyproject.toml
│   │   └── src/core/
│   │
│   ├── api/             # API service
│   │   ├── pyproject.toml
│   │   └── src/api/
│   │
│   └── worker/          # Background worker
│       ├── pyproject.toml
│       └── src/worker/
│
└── tools/               # Development tools
    └── scripts/

# pyproject.toml (workspace root)
[tool.uv.workspace]
members = ["packages/*"]

[tool.uv.sources]
core = { workspace = true }

python-project Extended Reference

This file preserves detailed material moved out of SKILL.md for progressive disclosure. Load it only when the current task needs the specific examples, commands, templates, or checklists below.

Moved content starts at: ## Testing.

Testing

# tests/conftest.py
import pytest
from httpx import ASGITransport, AsyncClient

from myapp.main import app


@pytest.fixture
async def client():
    async with AsyncClient(
        transport=ASGITransport(app=app),
        base_url="http://test",
    ) as client:
        yield client


# tests/test_user.py
import pytest


@pytest.mark.asyncio
async def test_create_user(client):
    response = await client.post(
        "/api/v1/users",
        json={"email": "test@example.com", "name": "Test User"},
    )
    assert response.status_code == 201
    data = response.json()
    assert data["email"] == "test@example.com"


@pytest.mark.asyncio
async def test_get_user_not_found(client):
    response = await client.get("/api/v1/users/00000000-0000-0000-0000-000000000000")
    assert response.status_code == 404

---

Makefile

.PHONY: dev test lint fmt check clean

# Run development server
dev:
	uv run uvicorn myapp.main:app --reload

# Run tests
test:
	uv run pytest

# Run tests with coverage
test-cov:
	uv run pytest --cov=myapp --cov-report=html

# Lint code
lint:
	uv run ruff check src tests

# Format code
fmt:
	uv run ruff format src tests
	uv run ruff check --fix src tests

# Type check
typecheck:
	uv run mypy src

# Run all checks
check: fmt lint typecheck test
	@echo "All checks passed!"

# Clean
clean:
	rm -rf .pytest_cache .mypy_cache .ruff_cache htmlcov .coverage
	find . -type d -name __pycache__ -exec rm -rf {} +

# Sync dependencies
sync:
	uv sync

# Upgrade dependencies
upgrade:
	uv lock --upgrade
	uv sync

---

Checklist

## Project Setup
- [ ] uv initialized with pyproject.toml
- [ ] .python-version set (3.12+)
- [ ] src/ layout structure
- [ ] Ruff configured
- [ ] mypy strict mode

## Architecture
- [ ] Pydantic models for validation
- [ ] Services for business logic
- [ ] Repositories for data access
- [ ] Custom exceptions
- [ ] Dependency injection

## Quality
- [ ] pytest with pytest-asyncio
- [ ] Type hints everywhere
- [ ] Structured logging
- [ ] Error handling middleware

## CI
- [ ] ruff check
- [ ] ruff format --check
- [ ] mypy
- [ ] pytest

---

See Also

• reference/architecture.md — Project structure patterns
• reference/tech-stack.md — Tool comparisons
• reference/patterns.md — Python design patterns

Python Design Patterns

Dependency Injection (FastAPI)

FastAPI's Depends for clean dependency management.

Basic Dependencies

# src/myapp/api/deps.py
from typing import Annotated

from fastapi import Depends
from sqlalchemy.ext.asyncio import AsyncSession

from myapp.db.session import async_session
from myapp.repositories.user import UserRepository
from myapp.services.user import UserService


async def get_session() -> AsyncSession:
    async with async_session() as session:
        yield session


async def get_user_repository(
    session: Annotated[AsyncSession, Depends(get_session)],
) -> UserRepository:
    return UserRepository(session)


async def get_user_service(
    repo: Annotated[UserRepository, Depends(get_user_repository)],
) -> UserService:
    return UserService(repo)


# Type aliases for cleaner signatures
SessionDep = Annotated[AsyncSession, Depends(get_session)]
UserServiceDep = Annotated[UserService, Depends(get_user_service)]

Using Dependencies

# src/myapp/api/routes/user.py
from fastapi import APIRouter

from myapp.api.deps import UserServiceDep
from myapp.models.user import User, UserCreate

router = APIRouter()


@router.post("/users", response_model=User)
async def create_user(
    data: UserCreate,
    service: UserServiceDep,
):
    return await service.create(data)

---

Repository Pattern

Abstract data access for testability.

Base Repository

# src/myapp/repositories/base.py
from typing import Generic, TypeVar
from uuid import UUID

from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from myapp.db.models import Base

ModelT = TypeVar("ModelT", bound=Base)


class BaseRepository(Generic[ModelT]):
    model: type[ModelT]

    def __init__(self, session: AsyncSession):
        self.session = session

    async def get(self, id: UUID) -> ModelT | None:
        return await self.session.get(self.model, id)

    async def get_all(self, *, skip: int = 0, limit: int = 100) -> list[ModelT]:
        result = await self.session.execute(
            select(self.model).offset(skip).limit(limit)
        )
        return list(result.scalars().all())

    async def create(self, obj: ModelT) -> ModelT:
        self.session.add(obj)
        await self.session.commit()
        await self.session.refresh(obj)
        return obj

    async def delete(self, obj: ModelT) -> None:
        await self.session.delete(obj)
        await self.session.commit()

Concrete Repository

# src/myapp/repositories/user.py
from sqlalchemy import select

from myapp.db.models import UserModel
from myapp.repositories.base import BaseRepository


class UserRepository(BaseRepository[UserModel]):
    model = UserModel

    async def get_by_email(self, email: str) -> UserModel | None:
        result = await self.session.execute(
            select(UserModel).where(UserModel.email == email)
        )
        return result.scalar_one_or_none()

---

Service Pattern

Business logic layer, framework-agnostic.

# src/myapp/services/user.py
from uuid import UUID

from myapp.core.exceptions import ConflictError, NotFoundError
from myapp.db.models import UserModel
from myapp.models.user import UserCreate, UserUpdate
from myapp.repositories.user import UserRepository


class UserService:
    def __init__(self, repo: UserRepository):
        self.repo = repo

    async def get(self, id: UUID) -> UserModel:
        user = await self.repo.get(id)
        if not user:
            raise NotFoundError("user", str(id))
        return user

    async def create(self, data: UserCreate) -> UserModel:
        # Business rule: email must be unique
        existing = await self.repo.get_by_email(data.email)
        if existing:
            raise ConflictError("email already exists")

        user = UserModel(
            email=data.email,
            name=data.name,
        )
        return await self.repo.create(user)

    async def update(self, id: UUID, data: UserUpdate) -> UserModel:
        user = await self.get(id)

        # Check email uniqueness if updating
        if data.email and data.email != user.email:
            existing = await self.repo.get_by_email(data.email)
            if existing:
                raise ConflictError("email already exists")

        # Update fields
        for field, value in data.model_dump(exclude_unset=True).items():
            setattr(user, field, value)

        await self.repo.session.commit()
        await self.repo.session.refresh(user)
        return user

    async def delete(self, id: UUID) -> None:
        user = await self.get(id)
        await self.repo.delete(user)

---

Custom Exceptions

Structured error handling.

# src/myapp/core/exceptions.py
from typing import Any


class AppError(Exception):
    """Base application error."""

    def __init__(
        self,
        message: str,
        code: str,
        status_code: int = 500,
        context: dict[str, Any] | None = None,
    ):
        self.message = message
        self.code = code
        self.status_code = status_code
        self.context = context or {}
        super().__init__(message)


class NotFoundError(AppError):
    def __init__(self, resource: str, id: str):
        super().__init__(
            message=f"{resource} not found: {id}",
            code="NOT_FOUND",
            status_code=404,
            context={"resource": resource, "id": id},
        )


class ConflictError(AppError):
    def __init__(self, message: str):
        super().__init__(
            message=message,
            code="CONFLICT",
            status_code=409,
        )


class ValidationError(AppError):
    def __init__(self, message: str, errors: list[dict] | None = None):
        super().__init__(
            message=message,
            code="VALIDATION_ERROR",
            status_code=400,
            context={"errors": errors or []},
        )


class UnauthorizedError(AppError):
    def __init__(self, message: str = "unauthorized"):
        super().__init__(
            message=message,
            code="UNAUTHORIZED",
            status_code=401,
        )

Exception Handler

# src/myapp/api/middleware.py
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse

from myapp.core.exceptions import AppError


def setup_exception_handlers(app: FastAPI) -> None:
    @app.exception_handler(AppError)
    async def app_error_handler(request: Request, exc: AppError) -> JSONResponse:
        return JSONResponse(
            status_code=exc.status_code,
            content={
                "error": {
                    "message": exc.message,
                    "code": exc.code,
                    **exc.context,
                }
            },
        )

---

Pydantic Patterns

Base Models

# src/myapp/models/base.py
from datetime import datetime
from uuid import UUID

from pydantic import BaseModel, ConfigDict


class AppModel(BaseModel):
    """Base model with common config."""

    model_config = ConfigDict(
        from_attributes=True,
        populate_by_name=True,
    )


class TimestampMixin(BaseModel):
    created_at: datetime
    updated_at: datetime


class IDModel(BaseModel):
    id: UUID

Request/Response Models

# src/myapp/models/user.py
from myapp.models.base import AppModel, IDModel, TimestampMixin


class UserBase(AppModel):
    email: str
    name: str


class UserCreate(UserBase):
    """Request model for creating user."""
    pass


class UserUpdate(AppModel):
    """Request model for updating user (all fields optional)."""
    email: str | None = None
    name: str | None = None


class User(UserBase, IDModel, TimestampMixin):
    """Response model with all fields."""
    pass

Validation

from pydantic import BaseModel, EmailStr, Field, field_validator


class UserCreate(BaseModel):
    email: EmailStr
    name: str = Field(min_length=2, max_length=100)
    password: str = Field(min_length=8)

    @field_validator("name")
    @classmethod
    def normalize_name(cls, v: str) -> str:
        return v.strip().title()

    @field_validator("password")
    @classmethod
    def validate_password(cls, v: str) -> str:
        if not any(c.isupper() for c in v):
            raise ValueError("must contain uppercase letter")
        if not any(c.isdigit() for c in v):
            raise ValueError("must contain digit")
        return v

---

Async Patterns

Async Context Manager

from contextlib import asynccontextmanager
from typing import AsyncGenerator

import httpx


@asynccontextmanager
async def get_http_client() -> AsyncGenerator[httpx.AsyncClient, None]:
    async with httpx.AsyncClient(timeout=30.0) as client:
        yield client


# Usage
async def fetch_data(url: str) -> dict:
    async with get_http_client() as client:
        response = await client.get(url)
        return response.json()

Concurrent Tasks

import asyncio


async def fetch_all(urls: list[str]) -> list[dict]:
    async with httpx.AsyncClient() as client:
        tasks = [client.get(url) for url in urls]
        responses = await asyncio.gather(*tasks)
        return [r.json() for r in responses]

TaskGroup (Python 3.11+)

import asyncio


async def process_items(items: list[str]) -> list[str]:
    results = []

    async with asyncio.TaskGroup() as tg:
        for item in items:
            tg.create_task(process_one(item, results))

    return results

Semaphore for Rate Limiting

import asyncio

import httpx


async def fetch_with_limit(urls: list[str], max_concurrent: int = 10) -> list[dict]:
    semaphore = asyncio.Semaphore(max_concurrent)

    async def fetch_one(url: str) -> dict:
        async with semaphore:
            async with httpx.AsyncClient() as client:
                response = await client.get(url)
                return response.json()

    return await asyncio.gather(*[fetch_one(url) for url in urls])

---

Factory Pattern

For creating objects with complex initialization.

# src/myapp/core/factories.py
from myapp.config import settings
from myapp.db.session import async_session
from myapp.repositories.user import UserRepository
from myapp.services.user import UserService


class ServiceFactory:
    @staticmethod
    async def create_user_service() -> UserService:
        async with async_session() as session:
            repo = UserRepository(session)
            return UserService(repo)


# Or using dependency injection
def create_user_service(repo: UserRepository) -> UserService:
    return UserService(repo)

---

Protocol Pattern (Structural Typing)

For interfaces without inheritance.

from typing import Protocol
from uuid import UUID


class UserRepositoryProtocol(Protocol):
    async def get(self, id: UUID) -> UserModel | None: ...
    async def get_by_email(self, email: str) -> UserModel | None: ...
    async def create(self, obj: UserModel) -> UserModel: ...
    async def delete(self, obj: UserModel) -> None: ...


class UserService:
    def __init__(self, repo: UserRepositoryProtocol):
        self.repo = repo


# Any class with these methods works, no inheritance needed
class InMemoryUserRepository:
    async def get(self, id: UUID) -> UserModel | None: ...
    async def get_by_email(self, email: str) -> UserModel | None: ...
    async def create(self, obj: UserModel) -> UserModel: ...
    async def delete(self, obj: UserModel) -> None: ...


# Works without explicitly implementing the protocol
service = UserService(InMemoryUserRepository())

---

Unit of Work Pattern

Transaction management across repositories.

# src/myapp/core/uow.py
from sqlalchemy.ext.asyncio import AsyncSession

from myapp.repositories.user import UserRepository


class UnitOfWork:
    def __init__(self, session: AsyncSession):
        self.session = session
        self.users = UserRepository(session)

    async def commit(self) -> None:
        await self.session.commit()

    async def rollback(self) -> None:
        await self.session.rollback()

    async def __aenter__(self) -> "UnitOfWork":
        return self

    async def __aexit__(self, exc_type, exc_val, exc_tb) -> None:
        if exc_type:
            await self.rollback()
        await self.session.close()


# Usage
async def transfer_credits(from_id: UUID, to_id: UUID, amount: int) -> None:
    async with UnitOfWork(session) as uow:
        from_user = await uow.users.get(from_id)
        to_user = await uow.users.get(to_id)

        from_user.credits -= amount
        to_user.credits += amount

        await uow.commit()

---

Summary Table

Python Tech Stack

Version Strategy

Always use latest. Never pin versions in templates.

[project]
dependencies = [
    "fastapi",    # uv resolves latest
    "pydantic",
]

• uv add fetches latest compatible versions
• uv.lock ensures reproducible builds
• uv lock --upgrade updates all dependencies

---

Package Management

uv (Recommended)

Rust-based, 10-100x faster than pip. Replaces pip, pip-tools, pyenv, virtualenv.

# Install
curl -LsSf https://astral.sh/uv/install.sh | sh

# Create project
uv init myapp
cd myapp

# Set Python version (auto-installs)
echo "3.12" > .python-version

# Add dependencies
uv add fastapi pydantic
uv add --dev pytest ruff

# Run commands
uv run python script.py
uv run pytest

# Sync environment
uv sync

Poetry

Mature, stable, good for library publishing.

poetry new myapp
poetry add fastapi pydantic
poetry add --group dev pytest ruff
poetry run pytest

Comparison

Recommendation: uv for new projects, Poetry for libraries.

---

Linting & Formatting

Ruff (Recommended)

Replaces Flake8 + Black + isort. 100-200x faster.

# pyproject.toml
[tool.ruff]
line-length = 100
target-version = "py312"

[tool.ruff.lint]
select = [
    "E",    # pycodestyle errors
    "F",    # pyflakes
    "I",    # isort
    "UP",   # pyupgrade
    "B",    # flake8-bugbear
    "SIM",  # flake8-simplify
    "RUF",  # ruff-specific
]
ignore = ["E501"]  # line too long (handled by formatter)

[tool.ruff.lint.isort]
known-first-party = ["myapp"]

[tool.ruff.format]
quote-style = "double"

# Lint
uv run ruff check src tests

# Format
uv run ruff format src tests

# Fix auto-fixable issues
uv run ruff check --fix src tests

Type Checking: mypy

[tool.mypy]
strict = true
python_version = "3.12"
warn_return_any = true
warn_unused_ignores = true
disallow_untyped_defs = true

[[tool.mypy.overrides]]
module = "tests.*"
disallow_untyped_defs = false

uv run mypy src

---

Web Frameworks

FastAPI (Recommended for APIs)

Modern, async, automatic OpenAPI docs.

from fastapi import FastAPI, Depends
from pydantic import BaseModel

app = FastAPI()


class User(BaseModel):
    name: str
    email: str


@app.post("/users")
async def create_user(user: User):
    return user

Pros:

• Automatic validation (Pydantic)
• Auto-generated docs (Swagger/ReDoc)
• Dependency injection
• Async native

Django (Full-stack)

Batteries included, admin panel, ORM.

# For full web apps with templates, auth, admin
# Not recommended for pure APIs

Flask

Minimal, flexible, sync by default.

# For simple apps or when you need maximum flexibility
# Consider FastAPI for new projects

Comparison

---

Data Validation

Pydantic v2 (Recommended)

Runtime validation with type hints. Rust core for speed.

from datetime import datetime
from uuid import UUID

from pydantic import BaseModel, EmailStr, Field, field_validator


class UserCreate(BaseModel):
    email: EmailStr
    name: str = Field(min_length=2, max_length=100)
    age: int = Field(ge=0, le=150)

    @field_validator("name")
    @classmethod
    def name_must_not_be_empty(cls, v: str) -> str:
        if not v.strip():
            raise ValueError("name cannot be empty")
        return v.strip()


class User(UserCreate):
    id: UUID
    created_at: datetime

    model_config = {"from_attributes": True}  # For ORM

Pydantic Settings

Environment configuration.

from pydantic_settings import BaseSettings


class Settings(BaseSettings):
    database_url: str
    debug: bool = False

    model_config = {"env_file": ".env"}

---

Database

SQLAlchemy 2.0 + asyncpg (Recommended)

Async support, type hints, mature.

from sqlalchemy import String
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column


class Base(DeclarativeBase):
    pass


class User(Base):
    __tablename__ = "users"

    id: Mapped[int] = mapped_column(primary_key=True)
    email: Mapped[str] = mapped_column(String(255), unique=True)
    name: Mapped[str] = mapped_column(String(100))


# Async query
async def get_user(session: AsyncSession, id: int) -> User | None:
    return await session.get(User, id)

SQLModel

Pydantic + SQLAlchemy combined (by FastAPI creator).

from sqlmodel import Field, SQLModel


class User(SQLModel, table=True):
    id: int | None = Field(default=None, primary_key=True)
    email: str = Field(unique=True, index=True)
    name: str

Comparison

---

HTTP Client

httpx (Recommended)

Async support, HTTP/2, requests-compatible API.

import httpx


# Async
async def fetch_data():
    async with httpx.AsyncClient() as client:
        response = await client.get("https://api.example.com/data")
        return response.json()


# Sync
def fetch_data_sync():
    response = httpx.get("https://api.example.com/data")
    return response.json()

requests

Classic, sync-only.

import requests

response = requests.get("https://api.example.com/data")

---

Testing

pytest (Recommended)

Simple, powerful, rich plugin ecosystem.

import pytest


def test_add():
    assert 1 + 1 == 2


@pytest.fixture
def user_data():
    return {"name": "Test", "email": "test@example.com"}


def test_create_user(user_data):
    assert user_data["name"] == "Test"

pytest-asyncio

Async test support.

import pytest


@pytest.mark.asyncio
async def test_async_function():
    result = await some_async_function()
    assert result == expected

[tool.pytest.ini_options]
asyncio_mode = "auto"  # No need for @pytest.mark.asyncio

Key Plugins

[tool.uv]
dev-dependencies = [
    "pytest",
    "pytest-asyncio",    # Async tests
    "pytest-cov",        # Coverage
    "pytest-xdist",      # Parallel execution
    "httpx",             # Async test client
]

---

Logging

structlog (Recommended)

Structured logging, great for production.

import structlog

structlog.configure(
    processors=[
        structlog.stdlib.add_log_level,
        structlog.processors.TimeStamper(fmt="iso"),
        structlog.processors.JSONRenderer(),
    ],
)

logger = structlog.get_logger()

logger.info("user_created", user_id=123, email="test@example.com")
# {"event": "user_created", "user_id": 123, "email": "test@example.com", "level": "info", "timestamp": "..."}

loguru

Simple API, colorful output.

from loguru import logger

logger.info("Processing {count} items", count=10)

---

CLI Tools

Typer (Recommended)

Click-based, type hints for arguments.

import typer

app = typer.Typer()


@app.command()
def hello(name: str, count: int = 1):
    for _ in range(count):
        print(f"Hello {name}!")


if __name__ == "__main__":
    app()

Click

Lower-level, more control.

import click


@click.command()
@click.option("--name", required=True)
def hello(name):
    click.echo(f"Hello {name}!")

---

Complete Stack Example

# pyproject.toml
[project]
name = "myapp"
version = "0.1.0"
requires-python = ">=3.12"
dependencies = [
    # Web
    "fastapi",
    "uvicorn[standard]",

    # Validation
    "pydantic",
    "pydantic-settings",

    # Database
    "sqlalchemy[asyncio]",
    "asyncpg",
    "alembic",

    # HTTP
    "httpx",

    # Logging
    "structlog",

    # LLM
    "openai",  # For LiteLLM proxy
]

[tool.uv]
dev-dependencies = [
    # Testing
    "pytest",
    "pytest-asyncio",
    "pytest-cov",

    # Quality
    "ruff",
    "mypy",

    # Types
    "types-passlib",
]