Building a Production-Style CRUD API with FastAPI, SQLAlchemy 2.0 and PostgreSQL

From an empty folder to a Dockerized API, written for developers who already know how to program but are new to the Python/FastAPI world.

If you've built REST APIs in Java, C#, Node.js or Go, you already know 90% of what you need. What you probably don't know yet is how the Python ecosystem wants you to do things: virtual environments instead of a global install, type hints that actually do something, and a small set of libraries that fit together surprisingly well.

In this post we'll build a complete CRUD API for a bookstore using:

FastAPI for the web framework
Uvicorn as the server
Pydantic v2 for validation and serialization
SQLAlchemy 2.0 as the ORM
PostgreSQL as the database (with psycopg2 as the driver)
Docker & Docker Compose to run everything with a single command

I've tried to build this the way I'd structure a real service, not a throwaway demo: layered code, dependency injection, proper status codes, config from environment variables, and tests. Along the way I'll point out the decisions I make and why, because that's the part tutorials usually skip.

Full source code: everything in this post is in the accompanying project https://github.com/leeonscoding/fastapi-crud. Every snippet below is taken from it and was run against a real PostgreSQL instance.

What we're building
A quick word on Uvicorn
Setting up a virtual environment
Installing FastAPI and Uvicorn
Hello, World
Running the server
A quick word on Pydantic
Swagger UI: free interactive docs
Project structure
Installing SQLAlchemy and psycopg2
Configuration with environment variables
Database setup: engine, session and Base
The SQLAlchemy model
The Pydantic schemas
Converting between Pydantic and SQLAlchemy
The CRUD layer
Session dependency injection
The router: GET, POST, PUT, PATCH, DELETE
Wiring it all together in main.py
Testing the API
Writing the Dockerfile
Running the app and the database with Docker Compose
Gotchas I've hit (so you don't have to)
Where to go next

1. What we're building

A small books API with these endpoints:

Method	Path	Description
`GET`	`/`	Hello World
`GET`	`/health`	Health check
`GET`	`/books?skip=0&limit=10`	List books (paginated)
`GET`	`/books/{id}`	Get one book
`POST`	`/books`	Create a book
`PUT`	`/books/{id}`	Replace a book
`PATCH`	`/books/{id}`	Partially update a book
`DELETE`	`/books/{id}`	Delete a book

A book has a title, author, a unique isbn, an optional published_year, and a created_at timestamp set by the database.

Prerequisites: Python 3.10+ (I'm using 3.12), Docker Desktop or Docker Engine with the Compose plugin, and a terminal.

2. A quick word on Uvicorn

When you run a Flask or Django app in production, there's a protocol sitting between your code and the web server called WSGI (Web Server Gateway Interface). It's synchronous: one request occupies one worker until it finishes.

FastAPI is built on ASGI (Asynchronous Server Gateway Interface), the async-capable successor. An ASGI application (FastAPI) needs an ASGI server to actually listen on a socket, parse HTTP and hand requests to your code. That server is Uvicorn.

How it works, in short:

Uvicorn opens a socket and runs an asyncio event loop (using the fast uvloop implementation when available).
It parses incoming HTTP using httptools.
For each request it calls your ASGI app, which is just an async callable: await app(scope, receive, send).
FastAPI (via Starlette) routes the request to your function and sends the response back through send.

Two practical consequences worth knowing up front:

If you write async def endpoints, they run on the event loop. Never call blocking code (like a synchronous DB driver) inside them, or you stall every other request.
If you write plain def endpoints, FastAPI runs them in a thread pool, so blocking calls are safe.

We're using the synchronous SQLAlchemy engine with psycopg2, so all our endpoints are plain def. That's a perfectly good, simple choice and I'll come back to it in the last section.

3. Setting up a virtual environment

Python installs packages globally by default, which turns into dependency hell the moment two projects need different versions of the same library. A virtual environment is an isolated folder with its own Python interpreter and its own site-packages. Think of it as node_modules plus a private runtime.

Create a project folder and the environment:

mkdir fastapi-crud
cd fastapi-crud
python -m venv venv

On some Linux/macOS machines the command is python3. On Debian/Ubuntu you may first need sudo apt install python3-venv.

Activate

Linux / macOS:

source venv/bin/activate

Windows (Command Prompt):

venv\Scripts\activate.bat

Windows (PowerShell):

venv\Scripts\Activate.ps1

If PowerShell complains that running scripts is disabled, run this once for your user and try again: Set-ExecutionPolicy -Scope CurrentUser -ExecutionPolicy RemoteSigned

You'll know it worked because your prompt gets a (venv) prefix.

Deactivate

Same command everywhere:

deactivate

Remember to add venv/ to your .gitignore. You never commit the environment, only the requirements.txt that can recreate it.

4. Installing FastAPI and Uvicorn

With the environment active:

pip install fastapi "uvicorn[standard]"

The [standard] extra pulls in uvloop, httptools, watchfiles (for auto-reload) and a few other performance goodies. Without it Uvicorn still works, just slower.

pip installs the latest versions. I like to pin exact versions so the project builds the same way next year as it does today. We'll collect everything in requirements.txt as we go. This is the final file:

fastapi==0.115.6
uvicorn[standard]==0.34.0
sqlalchemy==2.0.36
psycopg2-binary==2.9.10
pydantic==2.10.4
pydantic-settings==2.7.0

From now on you can set up the whole thing with one command:

pip install -r requirements.txt

5. Hello, World

Create app/main.py (we'll create the app package properly in a moment):

from fastapi import FastAPI

app = FastAPI(title="Bookstore CRUD API")


@app.get("/")
def hello_world():
    return {"message": "Hello, World!"}

That's the whole app. The @app.get("/") decorator registers the function as the handler for GET /. The returned dict is automatically serialized to JSON.

6. Running the server

uvicorn app.main:app --reload

Read app.main:app as "in the module app/main.py, find the variable named app". The --reload flag restarts the server when you change code, which is great for development and should never be used in production.

Open http://127.0.0.1:8000/ and you should see:

{"message": "Hello, World!"}

7. A quick word on Pydantic

Pydantic is a data validation library built around Python type hints. You declare what your data should look like as a class, and Pydantic validates, converts and serializes it for you:

from pydantic import BaseModel, Field

class Book(BaseModel):
    title: str = Field(min_length=1)
    published_year: int | None = None

Book(title="Clean Code", published_year="2008")   # "2008" is coerced to int 2008
Book(title="")                                     # raises ValidationError

FastAPI leans on Pydantic heavily:

Request bodies are parsed and validated automatically. Invalid input returns a 422 response that describes exactly what was wrong, and you wrote zero validation code.
Response models filter and serialize your output, so you never accidentally leak a field like password_hash.
OpenAPI documentation is generated from the models.

We're using Pydantic v2 (Rust core, much faster than v1). If you find older tutorials using class Config: orm_mode = True or .dict(), those are v1 idioms. The v2 equivalents are model_config = ConfigDict(from_attributes=True) and .model_dump().

8. Swagger UI: free interactive docs

FastAPI generates an OpenAPI schema from your code and ships two UIs for it, with no extra setup:

URL	What it is
http://127.0.0.1:8000/docs	Swagger UI: interactive, lets you try every endpoint from the browser
http://127.0.0.1:8000/redoc	ReDoc: cleaner read-only docs
http://127.0.0.1:8000/openapi.json	The raw OpenAPI schema (feed it to client generators)

Open /docs, expand an endpoint, click Try it out, then Execute. Coming from other ecosystems, this is the part that tends to win people over. You'll use it constantly for the rest of this post.

9. Project structure

Putting everything in main.py works for a demo but doesn't scale. This is the layout we'll build:

fastapi-crud/
├── app/
│   ├── __init__.py
│   ├── main.py            # app creation, lifespan, router registration
│   ├── config.py          # settings from environment variables
│   ├── database.py        # engine, session factory, Base, get_db dependency
│   ├── models.py          # SQLAlchemy models (database tables)
│   ├── schemas.py         # Pydantic models (API contract)
│   ├── crud.py            # database operations
│   └── routers/
│       ├── __init__.py
│       └── books.py       # HTTP endpoints
├── tests/
│   ├── __init__.py
│   ├── conftest.py
│   └── test_books.py
├── .env.example
├── .dockerignore
├── .gitignore
├── Dockerfile
├── docker-compose.yml
├── requirements.txt
└── requirements-dev.txt

The idea is a clean separation of concerns:

routers know about HTTP (status codes, query params, error responses)
crud knows about the database
schemas define what goes in and out over the wire
models define what's stored

A router never writes SQL, and a CRUD function never raises an HTTPException. That separation makes each layer easy to test and change.

Create the empty packages now:

mkdir -p app/routers tests
touch app/__init__.py app/routers/__init__.py tests/__init__.py

10. Installing SQLAlchemy and psycopg2

pip install sqlalchemy psycopg2-binary pydantic-settings

SQLAlchemy is the ORM and SQL toolkit. We use the modern 2.0 style (typed Mapped[...] columns and select() statements), not the legacy session.query() API you'll still see in older posts.
psycopg2-binary is the PostgreSQL driver. SQLAlchemy needs a driver to actually talk to the database.
pydantic-settings loads typed configuration from environment variables.

About -binary: it ships precompiled so you don't need libpq headers and a compiler. It's perfect for development and Docker images like ours. The psycopg maintainers recommend building from source for some production setups, and psycopg 3 is the newer driver worth a look. For this tutorial, psycopg2-binary keeps things simple.

11. Configuration with environment variables

Hard-coding a database URL is how credentials end up in Git. Following the twelve-factor app approach, we read config from the environment. app/config.py:

from functools import lru_cache

from pydantic_settings import BaseSettings, SettingsConfigDict


class Settings(BaseSettings):
    """Application settings, loaded from environment variables or a .env file."""

    postgres_user: str
    postgres_password: str
    postgres_db: str
    postgres_host: str = "localhost"
    postgres_port: int = 5432

    model_config = SettingsConfigDict(env_file=".env", extra="ignore")

    @property
    def database_url(self) -> str:
        return (
            f"postgresql+psycopg2://{self.postgres_user}:{self.postgres_password}"
            f"@{self.postgres_host}:{self.postgres_port}/{self.postgres_db}"
        )


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

A few details worth noting:

Fields without defaults (postgres_user, ...) are required. If one is missing the app fails at startup with a clear error rather than at 3 a.m. on the first query.
Environment variable names are matched case-insensitively, so POSTGRES_USER fills postgres_user.
@lru_cache makes get_settings() build the settings once and reuse them.

Create a .env.example that you do commit, and copy it to .env (which you don't):

POSTGRES_USER=bookuser
POSTGRES_PASSWORD=change_me
POSTGRES_DB=bookstore
POSTGRES_HOST=localhost
POSTGRES_PORT=5432

cp .env.example .env        # Linux/macOS
copy .env.example .env      # Windows (cmd)

A database to develop against

The easiest way to get PostgreSQL locally is Docker. We'll write the full Compose file later in this post, but you can already start just the database from it:

docker compose up -d db

(If you'd rather install PostgreSQL natively, that works too. Just create a user and database matching your .env.)

12. Database setup: engine, session and Base

app/database.py holds the three SQLAlchemy building blocks:

from collections.abc import Generator

from sqlalchemy import create_engine
from sqlalchemy.orm import DeclarativeBase, Session, sessionmaker

from app.config import get_settings


class Base(DeclarativeBase):
    """Base class for all SQLAlchemy models."""


engine = create_engine(
    get_settings().database_url,
    pool_pre_ping=True,  # transparently recover from stale connections
)

SessionLocal = sessionmaker(bind=engine, autoflush=False, expire_on_commit=False)


def get_db() -> Generator[Session, None, None]:
    """FastAPI dependency: one DB session per request, always closed afterwards."""
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

Let's unpack each piece, because this is where people new to SQLAlchemy usually get confused.

Base is the parent class for all your table models. It keeps a registry (Base.metadata) of every table so SQLAlchemy can create them and map them to classes.

engine is the application-wide entry point to the database. It owns a connection pool, and it's created once, not per request. pool_pre_ping=True makes the pool test a connection before handing it out, which saves you from "server closed the connection unexpectedly" errors after the database restarts or an idle connection is dropped.

SessionLocal is a factory that produces Session objects. A session is a unit of work. It tracks the objects you load or modify and talks to the database through a pooled connection. Two settings I choose deliberately:

autoflush=False: SQLAlchemy won't silently emit SQL before each query. I prefer explicit behavior over surprising queries.
expire_on_commit=False: by default, after commit() every attribute on your objects is marked stale and gets reloaded on next access. In a web API we serialize the object after committing, so we'd trigger extra queries (or errors if the session is closed). Turning this off avoids that.

13. The SQLAlchemy model

app/models.py, a class that maps to the books table:

from datetime import datetime

from sqlalchemy import DateTime, Integer, String, func
from sqlalchemy.orm import Mapped, mapped_column

from app.database import Base


class Book(Base):
    __tablename__ = "books"

    id: Mapped[int] = mapped_column(Integer, primary_key=True, index=True)
    title: Mapped[str] = mapped_column(String(200), nullable=False)
    author: Mapped[str] = mapped_column(String(120), nullable=False)
    isbn: Mapped[str] = mapped_column(String(20), unique=True, nullable=False)
    published_year: Mapped[int | None] = mapped_column(Integer, nullable=True)
    created_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), server_default=func.now()
    )

This is the SQLAlchemy 2.0 declarative style: Mapped[int] is a real type annotation that your editor and type checker understand, and mapped_column carries the database-specific details. Mapped[int | None] means the column is nullable.

created_at uses server_default=func.now(), so PostgreSQL sets the timestamp, not Python. That keeps it correct no matter who inserts the row.

14. The Pydantic schemas

The SQLAlchemy model describes how data is stored. Pydantic schemas describe how it travels over the API. These are different concerns and deserve different classes. For example, clients should never send id or created_at when creating a book. app/schemas.py:

from datetime import datetime

from pydantic import BaseModel, ConfigDict, Field


class BookBase(BaseModel):
    title: str = Field(min_length=1, max_length=200)
    author: str = Field(min_length=1, max_length=120)
    isbn: str = Field(min_length=10, max_length=20, examples=["9780132350884"])
    published_year: int | None = Field(default=None, ge=1000, le=2100)


class BookCreate(BookBase):
    """Payload for POST and PUT (full replacement)."""


class BookUpdate(BaseModel):
    """Payload for PATCH: every field is optional."""

    title: str | None = Field(default=None, min_length=1, max_length=200)
    author: str | None = Field(default=None, min_length=1, max_length=120)
    isbn: str | None = Field(default=None, min_length=10, max_length=20)
    published_year: int | None = Field(default=None, ge=1000, le=2100)


class BookRead(BookBase):
    """Response model. from_attributes lets Pydantic read SQLAlchemy objects."""

    model_config = ConfigDict(from_attributes=True)

    id: int
    created_at: datetime

Three schemas for one resource might feel like boilerplate at first, but each has a different job:

BookCreate: what the client sends to create (or fully replace) a book.
BookUpdate: for PATCH, where every field is optional. We'll use it to update only what the client sent.
BookRead: what the API returns. It adds the server-generated id and created_at.

The Field(...) constraints (min_length, ge, le) mean bad data is rejected with a 422 before it ever reaches your code or your database.

15. Converting between Pydantic and SQLAlchemy

This is the question I get asked most by people coming from other stacks. The two model families don't inherit from each other, so you convert at the boundary. Fortunately, it's only a line each way.

Pydantic → SQLAlchemy (on the way in)

model_dump() turns a Pydantic model into a dict, and the dict is unpacked into the ORM constructor:

book = Book(**payload.model_dump())   # payload is a BookCreate

For partial updates we only want the fields the client actually sent. exclude_unset=True is the key:

for field, value in payload.model_dump(exclude_unset=True).items():
    setattr(book, field, value)

Why this matters: with PATCH {"author": "Uncle Bob"}, model_dump() would also return title=None, isbn=None, ..., and you'd overwrite good data with None. exclude_unset=True distinguishes "not sent" from "explicitly sent as null".

SQLAlchemy → Pydantic (on the way out)

Because BookRead has from_attributes=True, Pydantic can read attributes straight off an ORM object:

BookRead.model_validate(book)   # book is a SQLAlchemy Book

In a FastAPI endpoint you usually don't even write this. Declare response_model=BookRead and return the ORM object. FastAPI performs the conversion and filtering for you:

@router.get("/{book_id}", response_model=BookRead)
def read_book(book: Annotated[Book, Depends(get_book_or_404)]):
    return book   # SQLAlchemy object in, BookRead JSON out

Remember the rule of thumb: Pydantic models are for the API boundary, SQLAlchemy models are for persistence. Don't use one as the other.

16. The CRUD layer

app/crud.py contains every database operation, and nothing about HTTP:

from sqlalchemy import select
from sqlalchemy.orm import Session

from app.models import Book
from app.schemas import BookCreate, BookUpdate


def get_book(db: Session, book_id: int) -> Book | None:
    return db.get(Book, book_id)


def get_book_by_isbn(db: Session, isbn: str) -> Book | None:
    return db.scalars(select(Book).where(Book.isbn == isbn)).first()


def list_books(db: Session, skip: int = 0, limit: int = 10) -> list[Book]:
    stmt = select(Book).order_by(Book.id).offset(skip).limit(limit)
    return list(db.scalars(stmt).all())


def create_book(db: Session, payload: BookCreate) -> Book:
    # Pydantic -> SQLAlchemy
    book = Book(**payload.model_dump())
    db.add(book)
    db.commit()
    db.refresh(book)
    return book


def replace_book(db: Session, book: Book, payload: BookCreate) -> Book:
    for field, value in payload.model_dump().items():
        setattr(book, field, value)
    db.commit()
    db.refresh(book)
    return book


def update_book(db: Session, book: Book, payload: BookUpdate) -> Book:
    # exclude_unset => only touch fields the client actually sent
    for field, value in payload.model_dump(exclude_unset=True).items():
        setattr(book, field, value)
    db.commit()
    db.refresh(book)
    return book


def delete_book(db: Session, book: Book) -> None:
    db.delete(book)
    db.commit()

Notes on the SQLAlchemy 2.0 query style:

Operation	Code	SQL
Select by PK	`db.get(Book, book_id)`	`SELECT ... WHERE id = :id` (and uses the session's identity map first)
Select with filter	`select(Book).where(...)` + `db.scalars(...)`	`SELECT ... WHERE ...`
Pagination	`.offset(skip).limit(limit)`	`... OFFSET :skip LIMIT :limit`
Insert	`db.add(obj)` + `commit()`	`INSERT ...`
Update	mutate attributes + `commit()`	`UPDATE ... SET ... WHERE id = :id`
Delete	`db.delete(obj)` + `commit()`	`DELETE ... WHERE id = :id`

db.scalars(stmt) returns ORM objects directly, not Row tuples. This is the one you want 95% of the time.
db.refresh(book) re-reads the row after commit, so server-generated values (id, created_at) are populated on the object.
list_books has an explicit order_by. Without it, PostgreSQL doesn't guarantee row order, which makes pagination unreliable.
Updates don't need an explicit UPDATE statement. SQLAlchemy tracks which attributes changed (the unit of work pattern) and emits the minimal UPDATE when you commit.

17. Session dependency injection

Here's the get_db function from database.py again:

def get_db() -> Generator[Session, None, None]:
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

FastAPI's dependency injection system is one of its best features. Any function can declare what it needs with Depends(...), and FastAPI resolves and supplies it per request:

DbSession = Annotated[Session, Depends(get_db)]

@router.get("")
def list_books(db: DbSession): ...

Because get_db is a generator with yield, FastAPI runs the code before the yield before your endpoint, passes db in, and runs the finally block after the response is produced, whether the endpoint succeeded or raised. That guarantees:

exactly one session per request, so concurrent requests never share state
the connection is always returned to the pool, so no leaks
endpoints don't know or care how sessions are created

The same mechanism gives us a big testing win. In the test suite we simply swap get_db for one that points at a different database (see section 20), with no mocking frameworks involved.

I use Annotated[Session, Depends(get_db)] aliased to DbSession so I don't repeat the Depends boilerplate in every signature. This is the style the FastAPI docs recommend today.

18. The router: GET, POST, PUT, PATCH, DELETE

app/routers/books.py. First the imports, shared dependencies and helpers:

from typing import Annotated

from fastapi import APIRouter, Depends, HTTPException, Query, Response, status
from sqlalchemy.orm import Session

from app import crud
from app.database import get_db
from app.models import Book
from app.schemas import BookCreate, BookRead, BookUpdate

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

DbSession = Annotated[Session, Depends(get_db)]


def get_book_or_404(book_id: int, db: DbSession) -> Book:
    book = crud.get_book(db, book_id)
    if book is None:
        raise HTTPException(status.HTTP_404_NOT_FOUND, f"Book {book_id} not found")
    return book


def ensure_isbn_available(db: Session, isbn: str, current_id: int | None = None) -> None:
    existing = crud.get_book_by_isbn(db, isbn)
    if existing is not None and existing.id != current_id:
        raise HTTPException(
            status.HTTP_409_CONFLICT, f"A book with ISBN {isbn} already exists"
        )

get_book_or_404 is itself a dependency. Any endpoint that takes a {book_id} can declare book: Book = Depends(get_book_or_404) and receive a loaded Book, or a 404 is raised before the endpoint body even runs. Dependencies can depend on other dependencies (get_book_or_404 itself uses DbSession), and FastAPI caches get_db within a request, so they all share the same session.

ensure_isbn_available turns a would-be database IntegrityError into a friendly 409 Conflict. The current_id argument lets a book keep its own ISBN during an update.

A note on races: this check-then-insert has a tiny race window if two requests with the same ISBN arrive simultaneously. The unique=True constraint on the column is the real safety net. In production I'd also catch IntegrityError around the commit and translate it to a 409. I kept it simple here to keep the focus on the fundamentals.

GET: list and read

@router.get("", response_model=list[BookRead])
def list_books(
    db: DbSession,
    skip: Annotated[int, Query(ge=0)] = 0,
    limit: Annotated[int, Query(ge=1, le=100)] = 10,
):
    return crud.list_books(db, skip=skip, limit=limit)


@router.get("/{book_id}", response_model=BookRead)
def read_book(book: Annotated[Book, Depends(get_book_or_404)]):
    return book  # SQLAlchemy object -> BookRead (via from_attributes)

skip and limit are query parameters (anything in the function signature that isn't in the path or body). I validate them with Query(ge=0) and le=100 so nobody can request a million rows with ?limit=1000000.

POST: create

@router.post("", response_model=BookRead, status_code=status.HTTP_201_CREATED)
def create_book(payload: BookCreate, db: DbSession):
    ensure_isbn_available(db, payload.isbn)
    return crud.create_book(db, payload)

A parameter typed as a Pydantic model is read from the request body. Note the 201 Created status code: returning 200 from a create endpoint is a very common beginner mistake.

PUT and PATCH: replace vs. partial update

@router.put("/{book_id}", response_model=BookRead)
def replace_book(
    payload: BookCreate,
    book: Annotated[Book, Depends(get_book_or_404)],
    db: DbSession,
):
    ensure_isbn_available(db, payload.isbn, current_id=book.id)
    return crud.replace_book(db, book, payload)


@router.patch("/{book_id}", response_model=BookRead)
def update_book(
    payload: BookUpdate,
    book: Annotated[Book, Depends(get_book_or_404)],
    db: DbSession,
):
    if payload.isbn is not None:
        ensure_isbn_available(db, payload.isbn, current_id=book.id)
    return crud.update_book(db, book, payload)

You asked for PUT, and I've added PATCH because the distinction matters in real APIs:

PUT replaces the resource. The client sends the full representation (BookCreate), and any field omitted resets to its default (for example published_year becomes null).
PATCH modifies the resource. The client sends only what changed (BookUpdate).

DELETE

@router.delete("/{book_id}", status_code=status.HTTP_204_NO_CONTENT)
def delete_book(
    book: Annotated[Book, Depends(get_book_or_404)],
    db: DbSession,
):
    crud.delete_book(db, book)
    return Response(status_code=status.HTTP_204_NO_CONTENT)

204 No Content means success with an empty body. Returning an explicit Response guarantees FastAPI doesn't try to send a JSON null body with a 204.

19. Wiring it all together in `main.py`

Now we replace the simple Hello World file with the final app/main.py:

from contextlib import asynccontextmanager

from fastapi import FastAPI

from app import models  # noqa: F401  (registers models on Base.metadata)
from app.database import Base, engine
from app.routers import books


@asynccontextmanager
async def lifespan(app: FastAPI):
    # Fine for a tutorial. In production, use Alembic migrations instead.
    Base.metadata.create_all(bind=engine)
    yield


app = FastAPI(
    title="Bookstore CRUD API",
    description="A CRUD API built with FastAPI, SQLAlchemy 2.0 and PostgreSQL.",
    version="1.0.0",
    lifespan=lifespan,
)

app.include_router(books.router)


@app.get("/", tags=["root"])
def hello_world():
    return {"message": "Hello, World!"}


@app.get("/health", tags=["root"])
def health():
    return {"status": "ok"}

Two things to note:

from app import models looks unused, but importing it is what registers the Book class on Base.metadata. Without it, create_all would create no tables. This is a classic gotcha, and the # noqa comment tells linters it's intentional.
lifespan is FastAPI's startup/shutdown hook (it replaces the deprecated @app.on_event("startup")). Code before yield runs at startup. create_all creates any missing tables, which is convenient for a tutorial. In a real project use Alembic migrations so schema changes are versioned and repeatable. create_all never alters existing tables.

Run it:

uvicorn app.main:app --reload

Open http://127.0.0.1:8000/docs and try the whole API from Swagger UI. Or use curl:

# create
curl -X POST http://127.0.0.1:8000/books \
  -H "Content-Type: application/json" \
  -d '{"title":"Clean Code","author":"Robert C. Martin","isbn":"9780132350884","published_year":2008}'

# list
curl "http://127.0.0.1:8000/books?skip=0&limit=10"

# partial update
curl -X PATCH http://127.0.0.1:8000/books/1 \
  -H "Content-Type: application/json" \
  -d '{"author":"Uncle Bob"}'

# delete
curl -X DELETE http://127.0.0.1:8000/books/1

Try sending an empty title or a duplicate ISBN, and observe the 422 and 409 responses.

20. Testing the API

An API without tests is a prototype. Install the dev dependencies:

pip install -r requirements-dev.txt

requirements-dev.txt:

-r requirements.txt
pytest==8.3.4
httpx==0.28.1

(httpx is what FastAPI's TestClient uses under the hood.)

This is where dependency injection pays off. tests/conftest.py replaces get_db with a session bound to an in-memory SQLite database, so the tests are fast and need no running PostgreSQL:

import os

# Settings must exist before the app is imported.
os.environ.setdefault("POSTGRES_USER", "test")
os.environ.setdefault("POSTGRES_PASSWORD", "test")
os.environ.setdefault("POSTGRES_DB", "test")

import pytest
from fastapi.testclient import TestClient
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from sqlalchemy.pool import StaticPool

from app.database import Base, get_db
from app.main import app


@pytest.fixture()
def client():
    # In-memory SQLite keeps the tests fast and dependency-free.
    engine = create_engine(
        "sqlite://",
        connect_args={"check_same_thread": False},
        poolclass=StaticPool,
    )
    Base.metadata.create_all(bind=engine)
    TestingSession = sessionmaker(bind=engine, autoflush=False, expire_on_commit=False)

    def override_get_db():
        db = TestingSession()
        try:
            yield db
        finally:
            db.close()

    app.dependency_overrides[get_db] = override_get_db
    yield TestClient(app)
    app.dependency_overrides.clear()

A few of the tests from tests/test_books.py:

BOOK = {
    "title": "Clean Code",
    "author": "Robert C. Martin",
    "isbn": "9780132350884",
    "published_year": 2008,
}


def test_create_and_get_book(client):
    created = client.post("/books", json=BOOK)
    assert created.status_code == 201
    body = created.json()
    assert body["id"] == 1 and body["title"] == BOOK["title"]

    fetched = client.get(f"/books/{body['id']}")
    assert fetched.status_code == 200
    assert fetched.json()["isbn"] == BOOK["isbn"]


def test_duplicate_isbn_returns_409(client):
    client.post("/books", json=BOOK)
    assert client.post("/books", json=BOOK).status_code == 409


def test_patch_updates_only_sent_fields(client):
    book_id = client.post("/books", json=BOOK).json()["id"]
    res = client.patch(f"/books/{book_id}", json={"author": "Uncle Bob"})
    assert res.json()["author"] == "Uncle Bob"
    assert res.json()["title"] == BOOK["title"]

Run them:

pytest -q

.........                                                                [100%]
9 passed

Trade-off I'm making consciously: SQLite is not PostgreSQL. It's fine for this simple schema, but once you use Postgres-specific features (JSONB, arrays, ON CONFLICT, advisory locks...), run your tests against a real Postgres container instead.

21. Writing the Dockerfile

Now let's package the app. Dockerfile:

FROM python:3.12-slim

ENV PYTHONDONTWRITEBYTECODE=1 \
    PYTHONUNBUFFERED=1 \
    PIP_NO_CACHE_DIR=1

WORKDIR /code

# Install dependencies first so this layer is cached until requirements change.
COPY requirements.txt .
RUN pip install -r requirements.txt

COPY app ./app

# Run as a non-root user.
RUN useradd --create-home appuser
USER appuser

EXPOSE 8000

CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000"]

What's worth explaining here:

python:3.12-slim is much smaller than the full image, and has everything we need (thanks to psycopg2-binary, no compiler is required).
PYTHONUNBUFFERED=1 makes logs show up immediately in docker logs instead of being buffered.
Layer caching: we copy requirements.txt and install before copying the source. Docker caches each step, so editing your code doesn't re-download every dependency. This one habit saves minutes per build.
Non-root user: if someone exploits the app, they don't get root inside the container.
--host 0.0.0.0 is essential. Uvicorn's default of 127.0.0.1 only listens inside the container, so you could never reach it from outside. This is the #1 "my container runs but I can't open it" cause.
No --reload. That's a development flag.

Add a .dockerignore so the build context stays small and never includes your local environment or secrets:

venv/
.venv/
__pycache__/
*.pyc
.env
.git
.pytest_cache/
tests/

Build and sanity check the image on its own (you'll need the DB for it to fully start, which is what Compose gives us next):

docker build -t bookstore-api .

22. Running the app and the database with Docker Compose

A single container isn't useful without its database. Docker Compose lets us describe both services and run them together. docker-compose.yml:

services:
  db:
    image: postgres:16-alpine
    restart: unless-stopped
    environment:
      POSTGRES_USER: ${POSTGRES_USER}
      POSTGRES_PASSWORD: ${POSTGRES_PASSWORD}
      POSTGRES_DB: ${POSTGRES_DB}
    volumes:
      - postgres_data:/var/lib/postgresql/data
    ports:
      - "5432:5432"
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U ${POSTGRES_USER} -d ${POSTGRES_DB}"]
      interval: 5s
      timeout: 5s
      retries: 10

  api:
    build: .
    restart: unless-stopped
    environment:
      POSTGRES_USER: ${POSTGRES_USER}
      POSTGRES_PASSWORD: ${POSTGRES_PASSWORD}
      POSTGRES_DB: ${POSTGRES_DB}
      POSTGRES_HOST: db   # the service name is the hostname on the compose network
      POSTGRES_PORT: 5432
    ports:
      - "8000:8000"
    depends_on:
      db:
        condition: service_healthy

volumes:
  postgres_data:

The important ideas:

${POSTGRES_USER} is substituted from your .env file, which Compose reads automatically. One source of truth for credentials, nothing hard-coded.
POSTGRES_HOST: db: inside the Compose network each service is reachable by its service name. This is the second most common Docker mistake. localhost inside the api container refers to the api container itself, not the database. Our Settings class defaults the host to localhost for local development, and Compose overrides it to db. This is exactly why we externalized config.
volumes: postgres_data is a named volume, so your data survives docker compose down and container recreation.
healthcheck + depends_on: condition: service_healthy: a plain depends_on only waits for the database container to start, not for PostgreSQL to be ready to accept connections. Without the health check, the API often boots first and crashes with "connection refused". pg_isready solves it properly.
ports: "5432:5432" exposes Postgres to your host so you can connect with a GUI such as DBeaver or pgAdmin. In production you would remove this and leave the database reachable only on the internal network.

Run everything

docker compose up --build

Add -d to run in the background. Then:

API: http://localhost:8000
Swagger UI: http://localhost:8000/docs

Useful commands:

docker compose logs -f api      # follow API logs
docker compose ps               # see service status and health
docker compose down             # stop and remove containers (data is kept)
docker compose down -v          # ...and delete the database volume too

Heads-up: the database credentials are applied only the first time the volume is initialized. If you change POSTGRES_PASSWORD in .env afterwards and the app can't authenticate, run docker compose down -v to recreate the volume (this deletes your data).

23. Gotchas I've hit (so you don't have to)

Tables not created. You forgot to import models somewhere before create_all. The models must be imported so they register on Base.metadata.
DetachedInstanceError or lazy-load errors after the request. Usually caused by touching ORM objects after the session is closed. expire_on_commit=False plus returning data through response_model avoids most of these.
sqlalchemy.exc.NoSuchModuleError: ... postgres. The URL scheme must be postgresql:// (or postgresql+psycopg2://). The old postgres:// alias was removed in SQLAlchemy 1.4.
Container can't reach the database. You used localhost instead of the service name db, or you're missing the health check.
Can't reach the API from the host in Docker. Uvicorn is bound to 127.0.0.1. Use --host 0.0.0.0.
.dict() / orm_mode deprecation warnings. You copied Pydantic v1 code. Use .model_dump() and from_attributes=True.
Using async def with a synchronous DB driver. This blocks the event loop. Keep def endpoints with the sync engine, or move fully to async (below).
Committing .env. Add it to .gitignore on day one, and commit .env.example instead.

24. Where to go next

You now have a layered, tested, containerized CRUD API. If I were taking this toward production, my next steps would be:

Alembic migrations instead of create_all. Versioned, reviewable schema changes.
Async stack: create_async_engine, AsyncSession, and the asyncpg driver with async def endpoints. It pays off under high concurrency, at the cost of more complexity. For many internal services, the synchronous version here is plenty.
Authentication: OAuth2 + JWT using FastAPI's fastapi.security utilities.
Better error handling: catch IntegrityError around commits and add a global exception handler for consistent error payloads.
Pagination metadata: return total, next, previous (or use cursor-based pagination for large tables).
Structured logging, metrics and tracing (OpenTelemetry).
CI pipeline: run pytest, a linter (ruff) and a type checker (mypy/pyright), then build and push the Docker image.
Multi-stage Docker build and Gunicorn with Uvicorn workers if you need multiple processes on a single host.

Wrapping up

The thing I appreciate most about this stack is how little ceremony it needs. A type-annotated function becomes a validated, documented endpoint. Dependency injection gives you clean resource management and trivial test overrides. And SQLAlchemy 2.0's typed style finally makes the ORM feel like a first-class citizen in a modern Python codebase.

If you're coming from another language, the main mental shifts are: always use a virtual environment, separate your Pydantic API models from your SQLAlchemy persistence models, and let the framework's dependency system manage sessions. Get those three right and the rest falls into place.

If you found this helpful, or spot something I could do better, I'd love to hear from you. Happy coding! 🚀

Tech used: Python 3.12 · FastAPI 0.115 · Uvicorn 0.34 · Pydantic 2.10 · SQLAlchemy 2.0 · PostgreSQL 16 · psycopg2 2.9 · Docker Compose

Table of Contents