# 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.

## Table of Contents

1. [What we're building](#1-what-were-building)
2. [A quick word on Uvicorn](#2-a-quick-word-on-uvicorn)
3. [Setting up a virtual environment](#3-setting-up-a-virtual-environment)
4. [Installing FastAPI and Uvicorn](#4-installing-fastapi-and-uvicorn)
5. [Hello, World](#5-hello-world)
6. [Running the server](#6-running-the-server)
7. [A quick word on Pydantic](#7-a-quick-word-on-pydantic)
8. [Swagger UI: free interactive docs](#8-swagger-ui-free-interactive-docs)
9. [Project structure](#9-project-structure)
10. [Installing SQLAlchemy and psycopg2](#10-installing-sqlalchemy-and-psycopg2)
11. [Configuration with environment variables](#11-configuration-with-environment-variables)
12. [Database setup: engine, session and Base](#12-database-setup-engine-session-and-base)
13. [The SQLAlchemy model](#13-the-sqlalchemy-model)
14. [The Pydantic schemas](#14-the-pydantic-schemas)
15. [Converting between Pydantic and SQLAlchemy](#15-converting-between-pydantic-and-sqlalchemy)
16. [The CRUD layer](#16-the-crud-layer)
17. [Session dependency injection](#17-session-dependency-injection)
18. [The router: GET, POST, PUT, PATCH, DELETE](#18-the-router-get-post-put-patch-delete)
19. [Wiring it all together in `main.py`](#19-wiring-it-all-together-in-mainpy)
20. [Testing the API](#20-testing-the-api)
21. [Writing the Dockerfile](#21-writing-the-dockerfile)
22. [Running the app and the database with Docker Compose](#22-running-the-app-and-the-database-with-docker-compose)
23. [Gotchas I've hit (so you don't have to)](#23-gotchas-ive-hit-so-you-dont-have-to)
24. [Where to go next](#24-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:

1. Uvicorn opens a socket and runs an **asyncio event loop** (using the fast `uvloop` implementation when available).
2. It parses incoming HTTP using `httptools`.
3. For each request it calls your ASGI app, which is just an async callable: `await app(scope, receive, send)`.
4. 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:

```bash
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:**

```bash
source venv/bin/activate
```

**Windows (Command Prompt):**

```bat
venv\Scripts\activate.bat
```

**Windows (PowerShell):**

```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:

```bash
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:

```bash
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:

```text
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:

```bash
pip install -r requirements.txt
```

---

## 5\. Hello, World

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

```python
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

```bash
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:

```json
{"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:

```python
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:

```text
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:

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

---

## 10\. Installing SQLAlchemy and psycopg2

```bash
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](https://12factor.net/config) approach, we read config from the environment. `app/config.py`:

```python
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*):

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

```bash
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:

```bash
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:

```python
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:

```python
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`:

```python
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:

```python
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:

```python
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:

```python
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:

```python
@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:

```python
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:

```python
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:

```python
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:

```python
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

```python
@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

```python
@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

```python
@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

```python
@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`:

```python
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:

1. `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.
2. `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**](https://alembic.sqlalchemy.org/) **migrations** so schema changes are versioned and repeatable. `create_all` never alters existing tables.

Run it:

```bash
uvicorn app.main:app --reload
```

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

```bash
# 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:

```bash
pip install -r requirements-dev.txt
```

`requirements-dev.txt`:

```text
-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:

```python
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`:

```python
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:

```bash
pytest -q
```

```text
.........                                                                [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`:

```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:

```text
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):

```bash
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`:

```yaml
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

```bash
docker compose up --build
```

Add `-d` to run in the background. Then:

- API: <http://localhost:8000>
- Swagger UI: <http://localhost:8000/docs>

Useful commands:

```bash
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)

1. **Tables not created.** You forgot to import `models` somewhere before `create_all`. The models must be imported so they register on `Base.metadata`.
2. `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.
3. `sqlalchemy.exc.NoSuchModuleError: ... postgres`**.** The URL scheme must be `postgresql://` (or `postgresql+psycopg2://`). The old `postgres://` alias was removed in SQLAlchemy 1.4.
4. **Container can't reach the database.** You used `localhost` instead of the service name `db`, or you're missing the health check.
5. **Can't reach the API from the host in Docker.** Uvicorn is bound to `127.0.0.1`. Use `--host 0.0.0.0`.
6. `.dict()` **/** `orm_mode` **deprecation warnings.** You copied Pydantic v1 code. Use `.model_dump()` and `from_attributes=True`.
7. **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).
8. **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*