FastAPI + Pydantic + SQLAlchemy

This page shows how to combine FastAPI, Pydantic, and SQLAlchemy in a way that stays readable under real service growth. The core move is to keep request DTOs, service/use-case logic, sessions and transactions, ORM records, and response DTOs as explicit boundaries.

Quick takeaway: let routes own HTTP concerns, let services own business actions, let repositories own persistence details, and keep commits at the use-case boundary. Finish the API with DTOs, not raw ORM entities.

End-to-End Flow

The API boundary should use DTOs, while the inside uses services, repositories, and one shared session per use case.

Recommended Folder Layout

app/
  api/
    routes/
      users.py
  schemas/
    user.py
  services/
    user_service.py
  repositories/
    user_repository.py
  db/
    models.py
    session.py
  domain/
    errors.py
  main.py

Boundary Types in Code

from datetime import datetime
from uuid import UUID

from pydantic import BaseModel, ConfigDict, EmailStr, Field


class CreateUserRequest(BaseModel):
    email: EmailStr
    name: str = Field(min_length=1, max_length=80)


class UserResponse(BaseModel):
    model_config = ConfigDict(from_attributes=True)

    id: UUID
    email: EmailStr
    name: str
    created_at: datetime

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


class UserRecord(Base):
    __tablename__ = "users"

    id: Mapped[UUID] = mapped_column(primary_key=True)
    email: Mapped[str] = mapped_column(String(255), unique=True, index=True)
    name: Mapped[str] = mapped_column(String(80))

`CreateUserRequest` and `UserResponse` are API contracts. `UserRecord` is a persistence implementation detail. Keeping them separate makes both layers easier to change.

Let the Service Own the Transaction

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


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

    async def get_by_email(self, email: str) -> UserRecord | None:
        stmt = select(UserRecord).where(UserRecord.email == email)
        return await self.session.scalar(stmt)

    def add(self, record: UserRecord) -> None:
        self.session.add(record)


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

    async def create_user(self, payload: CreateUserRequest) -> UserResponse:
        async with self.session.begin():
            if await self.users.get_by_email(payload.email) is not None:
                raise DuplicateEmail(payload.email)

            record = UserRecord(
                email=payload.email,
                name=payload.name,
            )
            self.users.add(record)
            await self.session.flush()

            return UserResponse.model_validate(record)

Keep the Route Thin

from collections.abc import AsyncIterator

from fastapi import APIRouter, Depends, status
from sqlalchemy.ext.asyncio import AsyncSession

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


async def get_session() -> AsyncIterator[AsyncSession]:
    async with AsyncSessionFactory() as session:
        yield session


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


@router.post("", response_model=UserResponse, status_code=status.HTTP_201_CREATED)
async def create_user(
    payload: CreateUserRequest,
    service: UserService = Depends(get_user_service),
) -> UserResponse:
    return await service.create_user(payload)

Design Reads and Writes Differently

Write path

• Transaction boundaries matter most.
• Use flush() when you need generated values or early constraint feedback.
• Build a response DTO without relying on lazy loads.

Read path

• A commit is often unnecessary.
• Choose loading strategies explicitly.
• Shape the result for the API instead of leaking ORM graphs.

A Near-Ideal API Design Checklist

ORM stays internal

Public contracts are DTOs, not ORM entities. Persistence changes should not force API changes.

One use case, one transaction

Let each business action own one atomic boundary.

Map domain errors to HTTP

Define domain errors like `DuplicateEmail` first, then translate them to status codes in routes or exception handlers.

Stop N+1 during serialization

Load what the response needs before building the DTO. Do not let serializers trigger database access.

The abc.ABC-based use-case plus class-based UoW pattern is covered separately in Use Case + UoW + ABC.

Avoid These Patterns

• Routes that call session.add(), session.commit(), and UserResponse.model_validate() directly
• Repository methods that commit on their own
• Reusing one schema for request and response when the field roles differ
• Returning ORM objects and hoping the response model will sort everything out

Official Sources

• FastAPI SQL Databases Tutorial
• SQLAlchemy Session Basics
• SQLAlchemy AsyncIO Support