Python Backend Expert

Python Backend Expert

When to Use

Activate this skill when:

• Creating or modifying FastAPI route handlers (endpoints)
• Defining or updating Pydantic v2 request/response schemas
• Writing SQLAlchemy 2.0 async models, queries, or relationships
• Implementing the repository pattern for data access
• Writing service layer business logic
• Creating or running Alembic migrations
• Setting up dependency injection chains with Depends()
• Handling errors across the route/service/repository layers

Do NOT use this skill for:

• Writing tests for backend code (use pytest-patterns)
• FastAPI framework mechanics — middleware, WebSockets, OpenAPI customization, CORS, lifespan (use fastapi-patterns)
• Deployment or CI/CD pipeline configuration (use deployment-pipeline)
• API contract design or endpoint planning (use api-design-patterns)
• Architecture decisions or layer design (use system-architecture)

Instructions

Project Structure

app/
├── main.py              # FastAPI application factory
├── core/
│   ├── config.py        # pydantic-settings configuration
│   ├── database.py      # Async engine, session factory
│   └── security.py      # Password hashing, JWT utilities
├── models/              # SQLAlchemy ORM models
│   ├── __init__.py
│   ├── base.py          # Declarative base
│   └── user.py
├── schemas/             # Pydantic v2 schemas
│   ├── __init__.py
│   └── user.py
├── repositories/        # Data access layer
│   ├── __init__.py
│   └── user_repo.py
├── services/            # Business logic layer
│   ├── __init__.py
│   └── user_service.py
├── routes/              # FastAPI routers
│   ├── __init__.py
│   └── users.py
├── dependencies/        # Reusable Depends() providers
│   ├── __init__.py
│   └── auth.py
└── exceptions.py        # Domain exception classes

FastAPI Endpoint Pattern

Every endpoint follows this structure:

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

@router.post("/", response_model=UserResponse, status_code=status.HTTP_201_CREATED)
async def create_user(
    data: UserCreate,
    session: AsyncSession = Depends(get_async_session),
) -> UserResponse:
    service = UserService(session)
    try:
        user = await service.create_user(data)
        return UserResponse.model_validate(user)
    except ConflictError as e:
        raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail=str(e))


@router.get("/{user_id}", response_model=UserResponse)
async def get_user(
    user_id: int,
    session: AsyncSession = Depends(get_async_session),
) -> UserResponse:
    service = UserService(session)
    try:
        user = await service.get_user(user_id)
        return UserResponse.model_validate(user)
    except NotFoundError as e:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(e))

Rules:

• Routes handle HTTP concerns only: status codes, HTTPException, response formatting
• Routes call services, never repositories directly
• Use response_model for automatic response serialization and OpenAPI docs
• Use status.HTTP_* constants, not bare integers
• Use Depends() for session, auth, and service injection

Repository Pattern

Repositories encapsulate all database access:

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

from app.models.user import User


class UserRepository:
    def __init__(self, session: AsyncSession) -> None:
        self._session = session

    async def get_by_id(self, user_id: int) -> User | None:
        result = await self._session.execute(
            select(User).where(User.id == user_id)
        )
        return result.scalar_one_or_none()

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

    async def list_with_posts(
        self, *, offset: int = 0, limit: int = 20
    ) -> list[User]:
        result = await self._session.execute(
            select(User)
            .options(selectinload(User.posts))
            .offset(offset)
            .limit(limit)
        )
        return list(result.scalars().all())

    async def create(self, user: User) -> User:
        self._session.add(user)
        await self._session.flush()
        await self._session.refresh(user)
        return user

    async def update(self, user: User, **kwargs: object) -> User:
        for key, value in kwargs.items():
            setattr(user, key, value)
        await self._session.flush()
        await self._session.refresh(user)
        return user

    async def delete(self, user: User) -> None:
        await self._session.delete(user)
        await self._session.flush()

Rules:

• One repository per model (or aggregate root)
• Repositories return model instances or None — never HTTP responses
• No business logic in repositories
• Always flush() + refresh() after add() to get generated fields (id, timestamps)
• Use selectinload() for eager loading relationships in async context
• Never raise HTTPException from repositories

Service Layer Pattern

Services contain business logic and orchestrate repositories:

from app.exceptions import ConflictError, NotFoundError
from app.models.user import User
from app.repositories.user_repo import UserRepository
from app.schemas.user import UserCreate, UserPatch
from app.core.security import hash_password


class UserService:
    def __init__(self, session: AsyncSession) -> None:
        self.repo = UserRepository(session)

    async def create_user(self, data: UserCreate) -> User:
        # Business rule: email must be unique
        existing = await self.repo.get_by_email(data.email)
        if existing:
            raise ConflictError(f"Email {data.email} already registered")

        # Business logic: hash password before storing
        user = User(
            email=data.email,
            hashed_password=hash_password(data.password),
            display_name=data.display_name,
        )
        return await self.repo.create(user)

    async def get_user(self, user_id: int) -> User:
        user = await self.repo.get_by_id(user_id)
        if user is None:
            raise NotFoundError(f"User {user_id} not found")
        return user

    async def update_user(self, user_id: int, data: UserPatch) -> User:
        user = await self.get_user(user_id)
        update_fields = data.model_dump(exclude_unset=True)
        if "password" in update_fields:
            update_fields["hashed_password"] = hash_password(update_fields.pop("password"))
        return await self.repo.update(user, **update_fields)

Rules:

• Services raise domain exceptions (NotFoundError, ConflictError), NEVER HTTPException
• Services are the only place for business logic
• Services call repositories for data access, never run raw queries
• Services receive AsyncSession via constructor and create their own repository instances
• Services validate business rules before calling repositories

Domain Exceptions

Define a hierarchy of domain exceptions:

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

class NotFoundError(AppError):
    """Resource not found."""

class ConflictError(AppError):
    """Resource conflict (duplicate, version mismatch)."""

class ValidationError(AppError):
    """Business rule violation."""

class PermissionError(AppError):
    """Insufficient permissions."""

Register global exception handlers in the FastAPI app:

from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse

app = FastAPI()

@app.exception_handler(NotFoundError)
async def not_found_handler(request: Request, exc: NotFoundError) -> JSONResponse:
    return JSONResponse(status_code=404, content={"detail": str(exc), "code": "NOT_FOUND"})

@app.exception_handler(ConflictError)
async def conflict_handler(request: Request, exc: ConflictError) -> JSONResponse:
    return JSONResponse(status_code=409, content={"detail": str(exc), "code": "CONFLICT"})

This allows services to raise domain exceptions without knowing about HTTP, and routes don't need try/except blocks.

Pydantic v2 Schema Conventions

from datetime import datetime
from pydantic import BaseModel, ConfigDict, EmailStr, Field


class UserCreate(BaseModel):
    """POST request body — writable fields only, no id/timestamps."""
    email: EmailStr
    password: str = Field(min_length=8, max_length=128)
    display_name: str = Field(min_length=1, max_length=100)


class UserPatch(BaseModel):
    """PATCH request body — all fields Optional."""
    email: EmailStr | None = None
    password: str | None = Field(default=None, min_length=8, max_length=128)
    display_name: str | None = Field(default=None, min_length=1, max_length=100)


class UserResponse(BaseModel):
    """Response body — all fields including id and timestamps."""
    model_config = ConfigDict(from_attributes=True)

    id: int
    email: str
    display_name: str
    is_active: bool
    created_at: datetime
    updated_at: datetime

Key Pydantic v2 patterns:

• Use ConfigDict(from_attributes=True) instead of class Config: orm_mode = True
• Use model_validate() instead of from_orm()
• Use model_dump() instead of .dict()
• Use model_dump(exclude_unset=True) for PATCH to distinguish "not sent" from "set to null"
• Use Field() for validation constraints
• Use str | None syntax (Python 3.12+), not Optional[str]

Async Session Management

from collections.abc import AsyncGenerator
from sqlalchemy.ext.asyncio import (
    AsyncSession,
    async_sessionmaker,
    create_async_engine,
)

from app.core.config import settings

engine = create_async_engine(
    settings.database_url,
    echo=settings.debug,
    pool_size=5,
    max_overflow=10,
    pool_pre_ping=True,
)

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


async def get_async_session() -> AsyncGenerator[AsyncSession, None]:
    async with async_session_factory() as session:
        async with session.begin():
            yield session

Rules:

• expire_on_commit=False prevents detached instance errors after commit
• session.begin() context manager auto-commits on success, rolls back on exception
• One session per request via Depends(get_async_session)
• Never share sessions across concurrent tasks
• For background tasks, create a new session — never reuse the request session

SQLAlchemy 2.0 Model Pattern

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


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, index=True)
    hashed_password: Mapped[str] = mapped_column(String(255))
    display_name: Mapped[str] = mapped_column(String(100))
    is_active: Mapped[bool] = mapped_column(default=True)
    created_at: Mapped[datetime] = mapped_column(server_default=func.now())
    updated_at: Mapped[datetime] = mapped_column(
        server_default=func.now(), onupdate=func.now()
    )

    # Relationships — ALWAYS use selectin or joined for async
    posts: Mapped[list["Post"]] = relationship(
        back_populates="author", lazy="selectin"
    )

Rules:

• Use Mapped[type] annotations (SQLAlchemy 2.0 style)
• Use mapped_column() instead of Column()
• Set lazy="selectin" on relationships for async compatibility
• Use server_default for database-generated defaults
• Always include created_at and updated_at timestamps

Alembic Migration Workflow

# Generate migration from model changes
alembic revision --autogenerate -m "add_users_table"

# Review the generated migration file before applying

# Apply migration
alembic upgrade head

# Rollback one step
alembic downgrade -1

# Show current revision
alembic current

# Show migration history
alembic history

Migration naming convention:

# alembic/env.py
naming_convention = {
    "ix": "ix_%(column_0_label)s",
    "uq": "uq_%(table_name)s_%(column_0_name)s",
    "ck": "ck_%(table_name)s_%(constraint_name)s",
    "fk": "fk_%(table_name)s_%(column_0_name)s_%(referred_table_name)s",
    "pk": "pk_%(table_name)s",
}

Rules:

• Always review autogenerated migrations before applying
• Every migration must have a working downgrade() function
• One migration per logical schema change
• Test both upgrade and downgrade
• Use descriptive migration messages: "add_users_table", "add_email_index_to_users"

Dependency Injection Pattern

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

from app.core.database import get_async_session
from app.services.user_service import UserService


async def get_user_service(
    session: AsyncSession = Depends(get_async_session),
) -> UserService:
    return UserService(session)


# Chain dependencies for auth
async def get_current_user(
    token: str = Depends(oauth2_scheme),
    session: AsyncSession = Depends(get_async_session),
) -> User:
    user_id = decode_token(token)
    service = UserService(session)
    return await service.get_user(user_id)


async def require_admin(
    user: User = Depends(get_current_user),
) -> User:
    if user.role != "admin":
        raise HTTPException(status_code=403, detail="Admin required")
    return user

Examples

Complete Request Flow

A request to POST /users flows through all layers: 1. Route receives UserCreate (Pydantic validates the request body) 2. Route calls UserService.create_user(data) via Depends() 3. Service checks business rule (email uniqueness) via UserRepository.get_by_email() 4. Service hashes password, creates User model instance 5. Service calls UserRepository.create(user) to persist 6. Repository adds to session, flushes, refreshes to get generated fields 7. Route converts the ORM model to UserResponse via model_validate()

If the email is duplicate, the service raises ConflictError, the global exception handler returns 409 Conflict. No try/except needed in the route.

Edge Cases

• Detached instance errors: Always call flush() + refresh() after session.add(). Set expire_on_commit=False on the session factory.

• Async session in background tasks: Never reuse the request session. Create a new session:

  async def background_job():
      async with async_session_factory() as session:
          async with session.begin():
              # do work

• N+1 queries: Use selectinload() in repository queries for relationships that will be accessed. Set lazy="selectin" as the default on model relationships.

• Bulk operations: Use session.execute(insert(User).values(list_of_dicts)) for bulk inserts instead of adding one by one.

• Transaction spanning multiple services: Pass the same session to all services. The session's begin() context manager handles the transaction boundary.

• Pydantic v2 computed fields: Use @computed_field for derived values in response schemas. See references/pydantic-v2-migration.md.

See references/sqlalchemy-patterns.md for advanced query optimization patterns.

Pydantic v2 Migration Patterns

Migration guide from Pydantic v1 to v2. Use this reference when updating existing code or when encountering v1-style patterns.

---

Key API Changes

Configuration

# v1 (deprecated)
class UserResponse(BaseModel):
    class Config:
        orm_mode = True
        allow_population_by_field_name = True

# v2 (current)
from pydantic import ConfigDict

class UserResponse(BaseModel):
    model_config = ConfigDict(
        from_attributes=True,        # replaces orm_mode
        populate_by_name=True,       # replaces allow_population_by_field_name
        str_strip_whitespace=True,   # new in v2
    )

Model Methods

Field Definitions

# v1 (deprecated)
from pydantic import Field

class User(BaseModel):
    name: str = Field(..., min_length=1)  # ... means required
    age: Optional[int] = None

# v2 (current)
class User(BaseModel):
    name: str = Field(min_length=1)       # required by default (no ...)
    age: int | None = None                # use | None instead of Optional

Validators

# v1 (deprecated)
from pydantic import validator, root_validator

class User(BaseModel):
    email: str

    @validator("email")
    @classmethod
    def validate_email(cls, v):
        return v.lower()

    @root_validator
    @classmethod
    def validate_model(cls, values):
        return values

# v2 (current)
from pydantic import field_validator, model_validator

class User(BaseModel):
    email: str

    @field_validator("email")
    @classmethod
    def validate_email(cls, v: str) -> str:
        return v.lower()

    @model_validator(mode="after")
    def validate_model(self) -> "User":
        # self is the fully constructed model
        return self

Computed Fields (New in v2)

from pydantic import computed_field

class OrderResponse(BaseModel):
    subtotal_cents: int
    tax_cents: int

    @computed_field
    @property
    def total_cents(self) -> int:
        return self.subtotal_cents + self.tax_cents

---

Type Annotation Changes

# v1 style
from typing import Optional, List, Dict

class User(BaseModel):
    tags: List[str] = []
    metadata: Dict[str, str] = {}
    nickname: Optional[str] = None

# v2 style (Python 3.12+)
class User(BaseModel):
    tags: list[str] = []
    metadata: dict[str, str] = {}
    nickname: str | None = None

---

Strict Mode

Pydantic v2 introduces strict mode to prevent type coercion:

from pydantic import BaseModel, ConfigDict

class StrictUser(BaseModel):
    model_config = ConfigDict(strict=True)

    age: int
    name: str

# Without strict: StrictUser(age="25", name="Alice") → age=25 (coerced)
# With strict: StrictUser(age="25", name="Alice") → ValidationError

Per-field strict mode:

from pydantic import Field

class User(BaseModel):
    age: int = Field(strict=True)  # Only this field is strict
    name: str

---

Discriminated Unions (Improved in v2)

from typing import Annotated, Literal, Union
from pydantic import BaseModel, Discriminator, Tag

class Cat(BaseModel):
    pet_type: Literal["cat"]
    meow_volume: int

class Dog(BaseModel):
    pet_type: Literal["dog"]
    bark_volume: int

# v2 discriminated union
Pet = Annotated[
    Union[
        Annotated[Cat, Tag("cat")],
        Annotated[Dog, Tag("dog")],
    ],
    Discriminator("pet_type"),
]

class Owner(BaseModel):
    pet: Pet  # Automatically selects Cat or Dog based on pet_type

---

Common Migration Patterns

Pattern 1: ORM Model to Response

# v1
response = UserResponse.from_orm(user_model)

# v2
response = UserResponse.model_validate(user_model)

Pattern 2: Partial Update (PATCH)

# v1
update_data = patch_schema.dict(exclude_unset=True)

# v2
update_data = patch_schema.model_dump(exclude_unset=True)

Pattern 3: Response Serialization

# v1
return user.dict(exclude={"hashed_password"})

# v2
return user.model_dump(exclude={"hashed_password"})

Pattern 4: JSON Serialization

# v1
json_str = user.json()
user = User.parse_raw(json_str)

# v2
json_str = user.model_dump_json()
user = User.model_validate_json(json_str)

Pattern 5: Schema Copy with Update

# v1
updated = user.copy(update={"name": "New Name"})

# v2
updated = user.model_copy(update={"name": "New Name"})

---

Deprecated Features to Remove

SQLAlchemy 2.0 Advanced Patterns

Advanced query patterns, relationship loading, bulk operations, and session management for async SQLAlchemy 2.0 with FastAPI.

---

Relationship Loading Strategies

selectinload (Default for Async)

Loads related objects in a separate SELECT with an IN clause. Best for one-to-many and many-to-many relationships.

from sqlalchemy.orm import selectinload

# Load a user with all their posts
result = await session.execute(
    select(User)
    .where(User.id == user_id)
    .options(selectinload(User.posts))
)
user = result.scalar_one_or_none()
# user.posts is loaded — no lazy loading needed

When to use: Default choice for async. Good when you need the related objects and the collection is moderate-sized.

joinedload

Loads related objects in a single JOIN query. Best for many-to-one and one-to-one relationships.

from sqlalchemy.orm import joinedload

# Load posts with their authors in a single query
result = await session.execute(
    select(Post)
    .options(joinedload(Post.author))
    .limit(20)
)
posts = result.unique().scalars().all()
# post.author is loaded for each post

Important: Always call .unique() on the result when using joinedload with collections, as the JOIN can produce duplicate rows.

When to use: For to-one relationships where you always need the related object.

subqueryload

Loads related objects in a separate subquery. Similar to selectinload but uses a subquery instead of IN.

from sqlalchemy.orm import subqueryload

result = await session.execute(
    select(User)
    .options(subqueryload(User.posts))
)

When to use: When the parent query is complex and IN clause would be too large.

Nested Loading

Load relationships of relationships:

result = await session.execute(
    select(User)
    .options(
        selectinload(User.posts).selectinload(Post.comments)
    )
)

raiseload (Prevent Accidental Lazy Loads)

from sqlalchemy.orm import raiseload

result = await session.execute(
    select(User)
    .options(raiseload("*"))  # Raise error on any lazy load attempt
    .options(selectinload(User.posts))  # Explicitly load what you need
)

When to use: In development/testing to catch N+1 queries early.

---

Query Optimization

Selecting Specific Columns

# Only load the columns you need
result = await session.execute(
    select(User.id, User.email, User.display_name)
    .where(User.is_active == True)
)
rows = result.all()  # Returns tuples, not User instances

Aggregation

from sqlalchemy import func

# Count active users
result = await session.execute(
    select(func.count()).select_from(User).where(User.is_active == True)
)
count = result.scalar_one()

# Group by with count
result = await session.execute(
    select(User.role, func.count(User.id).label("count"))
    .group_by(User.role)
)
role_counts = result.all()

Exists Check

from sqlalchemy import exists

# Efficient existence check (doesn't load the row)
result = await session.execute(
    select(exists().where(User.email == email))
)
email_exists = result.scalar_one()

Pagination with Cursor

async def list_users_cursor(
    session: AsyncSession,
    *,
    after_id: int | None = None,
    limit: int = 20,
) -> tuple[list[User], bool]:
    query = select(User).order_by(User.id)

    if after_id is not None:
        query = query.where(User.id > after_id)

    # Fetch one extra to determine has_more
    query = query.limit(limit + 1)

    result = await session.execute(query)
    users = list(result.scalars().all())

    has_more = len(users) > limit
    if has_more:
        users = users[:limit]

    return users, has_more

---

Bulk Operations

Bulk Insert

from sqlalchemy import insert

# Insert many rows efficiently
users_data = [
    {"email": "a@example.com", "display_name": "Alice", "hashed_password": "..."},
    {"email": "b@example.com", "display_name": "Bob", "hashed_password": "..."},
]
await session.execute(insert(User), users_data)
await session.flush()

Bulk Update

from sqlalchemy import update

# Update many rows at once
await session.execute(
    update(User)
    .where(User.last_login < cutoff_date)
    .values(is_active=False)
)
await session.flush()

Bulk Delete

from sqlalchemy import delete

await session.execute(
    delete(User).where(User.is_active == False)
)
await session.flush()

---

Connection Pool Tuning

engine = create_async_engine(
    database_url,
    pool_size=5,          # Steady-state connections
    max_overflow=10,      # Additional connections under load
    pool_pre_ping=True,   # Verify connections before use
    pool_recycle=3600,    # Recycle connections after 1 hour
    pool_timeout=30,      # Wait time for available connection
    echo=False,           # Set True for SQL logging in development
)

Guidelines:

• pool_size: Set to expected concurrent database sessions (usually matches web worker count)
• max_overflow: Additional connections allowed above pool_size during traffic spikes
• pool_pre_ping: Always enable to handle database restarts gracefully
• pool_recycle: Set below the database's connection timeout (PostgreSQL default: 8 hours)

---

Async Session Patterns

Request-Scoped Session

async def get_async_session() -> AsyncGenerator[AsyncSession, None]:
    async with async_session_factory() as session:
        async with session.begin():
            yield session
            # Auto-commits on success, rolls back on exception

Background Task Session

async def send_welcome_email(user_id: int) -> None:
    """Background task — creates its own session."""
    async with async_session_factory() as session:
        async with session.begin():
            user = await session.get(User, user_id)
            if user:
                await email_service.send(user.email, "Welcome!")

Test Session with Rollback

@pytest.fixture
async def db_session(engine):
    async with engine.connect() as conn:
        await conn.begin()
        async_session = AsyncSession(bind=conn, expire_on_commit=False)

        yield async_session

        await async_session.close()
        await conn.rollback()

---

Index Strategy

from sqlalchemy import Index

class Order(Base):
    __tablename__ = "orders"

    id: Mapped[int] = mapped_column(primary_key=True)
    user_id: Mapped[int] = mapped_column(ForeignKey("users.id"), index=True)
    status: Mapped[str] = mapped_column(String(20))
    created_at: Mapped[datetime] = mapped_column(server_default=func.now())

    # Composite index for common query pattern
    __table_args__ = (
        Index("ix_orders_user_status", "user_id", "status"),
        Index(
            "ix_orders_active",
            "user_id", "created_at",
            postgresql_where=(status != "cancelled"),  # Partial index
        ),
    )