Chapter 10: PostgreSQL DB Setup and FastAPI with CRUD API

Let’s Start With a Real Problem

SQLite has been a fantastic learning partner — zero setup, one file, instant results. But if you tried to deploy your Chapter 8 or 9 project to a real hosting platform like Render, Railway, or AWS, you’d quickly hit a wall: most cloud platforms either don’t support writing to a local SQLite file at all (their filesystems reset on every deploy), or they strongly discourage it for anything beyond small side projects. SQLite simply wasn’t designed for multiple servers, high concurrency, or the kind of workloads production APIs deal with.

PostgreSQL is the database most real-world FastAPI teams reach for — powerful, open-source, and rock-solid at scale. In this chapter, you’ll set it up locally, connect it to FastAPI, and rebuild a complete CRUD API on top of it. The best part? Thanks to SQLAlchemy, almost none of your application code changes — proof that the layered structure we’ve been building since Chapter 5 really pays off.

---

What You’ll Learn

• Why PostgreSQL is the industry-standard choice for production APIs
• How to run PostgreSQL locally using Docker (the modern, fastest way — no manual installers)
• How to connect FastAPI to PostgreSQL using SQLAlchemy and the psycopg driver
• How little application code actually needs to change when switching databases
• How to build a complete CRUD API (Create, Read, Update, Delete) backed by PostgreSQL
• Theory + hands-on assignment questions to lock in what you’ve learned

💡 A note on secrets: In this chapter, we’ll keep the database connection string as a simple default setting to stay focused on PostgreSQL itself. In Chapter 13, we’ll properly move credentials like this into a .env file using python-dotenv — the correct way to handle secrets in a real project. For now, just know we’re intentionally simplifying this one piece.

---

Theory: Why PostgreSQL?

	SQLite	PostgreSQL
Setup	Zero — just a file	Needs a running server (or Docker container)
Concurrency	Limited — struggles with many simultaneous writes	Built for many concurrent connections
Data types	Basic	Rich — JSON columns, arrays, full-text search, and more
Best for	Learning, prototypes, small local tools	Production APIs, real applications, team projects
Hosting	Awkward on most cloud platforms	Supported everywhere (AWS RDS, Railway, Render, Supabase…)

The good news: because we’ve been using SQLAlchemy as an abstraction layer since Chapter 8, switching databases is mostly a configuration change, not a code rewrite. That’s the entire point of using an ORM.

---

Program 1: Connecting FastAPI to PostgreSQL

Program Objective

Set up a real PostgreSQL database using Docker, connect it to a small FastAPI project using SQLAlchemy, and build basic POST /users/register and GET /users/ endpoints — proving the connection works end to end.

Step-by-Step Implementation

Step 1: Start a PostgreSQL container using Docker

If you don’t have Docker installed, grab it from docker.com. Then run:

docker run --name fastapi-postgres -e POSTGRES_USER=fastapi_user -e POSTGRES_PASSWORD=fastapi_pass -e POSTGRES_DB=fastapi_db -p 5432:5432 -d postgres:16

This starts a real PostgreSQL 16 server on your machine in seconds, with:

• Username: fastapi_user
• Password: fastapi_pass
• Database name: fastapi_db
• Accessible at localhost:5432

💡 Why Docker instead of a native installer? It’s faster to set up, easy to completely remove (docker rm -f fastapi-postgres) without leaving anything behind on your system, and it’s exactly how most professional teams run local databases during development.

Step 2: Create the project

uv init postgres-connection-demo
cd postgres-connection-demo
uv add fastapi --extra standard
uv add pydantic-settings
uv add sqlalchemy
uv add "psycopg[binary]"

💡 psycopg (version 3) is the modern, officially recommended PostgreSQL driver for SQLAlchemy 2.0. The [binary] extra bundles a precompiled version, so you don’t need PostgreSQL’s development headers installed on your machine.

Step 3: Create the project structure

mkdir -p app/core app/db app/models app/schemas app/services app/routers
touch app/__init__.py app/core/__init__.py app/db/__init__.py app/models/__init__.py app/schemas/__init__.py app/services/__init__.py app/routers/__init__.py

postgres-connection-demo/
├── app/
│   ├── __init__.py
│   ├── main.py
│   ├── core/
│   │   ├── __init__.py
│   │   └── config.py
│   ├── db/
│   │   ├── __init__.py
│   │   └── database.py
│   ├── models/
│   │   ├── __init__.py
│   │   └── user.py
│   ├── schemas/
│   │   ├── __init__.py
│   │   └── user.py
│   ├── services/
│   │   ├── __init__.py
│   │   └── user_service.py
│   └── routers/
│       ├── __init__.py
│       └── users.py
├── pyproject.toml
└── uv.lock

Step 4: Create each file with the code below. Step 5: Run and test.

Complete Working Source Code

app/core/config.py

from pydantic_settings import BaseSettings


class Settings(BaseSettings):
    app_name: str = "Postgres Connection Demo"
    app_version: str = "1.0.0"
    database_url: str = "postgresql+psycopg://fastapi_user:fastapi_pass@localhost:5432/fastapi_db"


settings = Settings()

app/db/database.py

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

from app.core.config import settings

engine = create_engine(settings.database_url, pool_pre_ping=True)

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


class Base(DeclarativeBase):
    pass


def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

app/models/user.py

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

from app.db.database import Base


class User(Base):
    __tablename__ = "users"

    id: Mapped[int] = mapped_column(primary_key=True, index=True)
    username: Mapped[str] = mapped_column(String(30), unique=True, index=True)
    email: Mapped[str] = mapped_column(String(255), unique=True, index=True)
    hashed_password: Mapped[str] = mapped_column(String(255))
    is_active: Mapped[bool] = mapped_column(Boolean, default=True)

app/schemas/user.py

from typing import Annotated

from pydantic import BaseModel, ConfigDict, EmailStr, Field


class UserCreate(BaseModel):
    username: Annotated[str, Field(min_length=3, max_length=30)]
    email: Annotated[EmailStr, Field()]
    password: Annotated[str, Field(min_length=8, max_length=64)]


class UserOut(BaseModel):
    id: int
    username: str
    email: EmailStr
    is_active: bool

    model_config = ConfigDict(from_attributes=True)

app/services/user_service.py

from sqlalchemy.orm import Session

from app.models.user import User
from app.schemas.user import UserCreate


def create_user(db: Session, user_in: UserCreate) -> User:
    new_user = User(
        username=user_in.username,
        email=user_in.email,
        hashed_password=f"hashed_{user_in.password}",
    )
    db.add(new_user)
    db.commit()
    db.refresh(new_user)
    return new_user


def list_users(db: Session) -> list[User]:
    return db.query(User).all()

app/routers/users.py

from fastapi import APIRouter, Depends, status
from sqlalchemy.orm import Session

from app.db.database import get_db
from app.schemas.user import UserCreate, UserOut
from app.services.user_service import create_user, list_users

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


@router.post("/register", response_model=UserOut, status_code=status.HTTP_201_CREATED)
def register_user(user_in: UserCreate, db: Session = Depends(get_db)):
    return create_user(db, user_in)


@router.get("/", response_model=list[UserOut])
def get_all_users(db: Session = Depends(get_db)):
    return list_users(db)

app/main.py

from fastapi import FastAPI

from app.core.config import settings
from app.db.database import Base, engine
from app.models import user  # noqa: F401
from app.routers import users

Base.metadata.create_all(bind=engine)

app = FastAPI(title=settings.app_name, version=settings.app_version)

app.include_router(users.router)


@app.get("/", tags=["Root"])
def read_root():
    return {"message": f"Welcome to {settings.app_name}"}

Detailed Code Explanation

• The connection string is the star of this program: postgresql+psycopg://fastapi_user:fastapi_pass@localhost:5432/fastapi_db follows the pattern dialect+driver://username:password@host:port/database_name. Compare this to Chapter 8’s SQLite URL (sqlite:///./app.db) — that’s genuinely the only meaningful difference in our configuration.
• pool_pre_ping=True in create_engine(...) tells SQLAlchemy to quietly test each connection before using it, and transparently reconnect if it’s gone stale — a small but important reliability setting for real network-based databases (SQLite, being a local file, never needed this).
• Notice we removed connect_args={"check_same_thread": False} — that setting was an SQLite-specific workaround from Chapter 8 and doesn’t apply to PostgreSQL at all.
• Every other file — models/user.py, schemas/user.py, services/user_service.py, routers/users.py — is completely unchanged from the SQLite version in Chapter 8. This is exactly the payoff of using SQLAlchemy: your models, business logic, and API routes don’t know or care which actual database engine is running underneath them.
• Base.metadata.create_all(bind=engine) behaves identically too — it inspects your models and creates the corresponding tables, this time inside your PostgreSQL database instead of a SQLite file.

Execution & Testing

Make sure your Docker container from Step 1 is running (docker ps should show fastapi-postgres), then:

uv run fastapi dev app/main.py

1. Open http://127.0.0.1:8000/docs and call POST /users/register with a sample user.
2. Call GET /users/ and confirm the user comes back.
3. Verify it’s really PostgreSQL: run docker exec -it fastapi-postgres psql -U fastapi_user -d fastapi_db -c "SELECT * FROM users;" in your terminal — you should see your user’s row printed directly from the PostgreSQL database itself, not a local file.

---

Program 2: A Complete CRUD API on PostgreSQL

Program Objective

Build a full, standalone CRUD API — Create, Read (all + one), Update (PATCH), and Delete — for a Product resource, backed entirely by PostgreSQL, consolidating everything from Chapters 7, 8, and 9 into one clean, production-shaped project.

Step-by-Step Implementation

Step 1: Start a fresh PostgreSQL container

docker run --name fastapi-postgres-crud -e POSTGRES_USER=fastapi_user -e POSTGRES_PASSWORD=fastapi_pass -e POSTGRES_DB=products_db -p 5433:5432 -d postgres:16

💡 We used port 5433 (instead of 5432) and a new container name, so this can run alongside Program 1’s container without conflicts if you still have it running.

Step 2: Create the project

uv init postgres-crud-demo
cd postgres-crud-demo
uv add fastapi --extra standard
uv add pydantic-settings
uv add sqlalchemy
uv add "psycopg[binary]"

Step 3: Create the project structure

mkdir -p app/core app/db app/models app/schemas app/services app/routers
touch app/__init__.py app/core/__init__.py app/db/__init__.py app/models/__init__.py app/schemas/__init__.py app/services/__init__.py app/routers/__init__.py

postgres-crud-demo/
├── app/
│   ├── __init__.py
│   ├── main.py
│   ├── core/
│   │   ├── __init__.py
│   │   ├── config.py
│   │   └── exceptions.py
│   ├── db/
│   │   ├── __init__.py
│   │   └── database.py
│   ├── models/
│   │   ├── __init__.py
│   │   └── product.py
│   ├── schemas/
│   │   ├── __init__.py
│   │   └── product.py
│   ├── services/
│   │   ├── __init__.py
│   │   └── product_service.py
│   └── routers/
│       ├── __init__.py
│       └── products.py
├── pyproject.toml
└── uv.lock

Step 4: Create each file with the code below. Step 5: Run and test.

Complete Working Source Code

app/core/config.py

from pydantic_settings import BaseSettings


class Settings(BaseSettings):
    app_name: str = "Postgres CRUD Demo"
    app_version: str = "1.0.0"
    database_url: str = "postgresql+psycopg://fastapi_user:fastapi_pass@localhost:5433/products_db"


settings = Settings()

app/core/exceptions.py

class ProductNotFoundError(Exception):
    """Raised when a requested product does not exist."""

    def __init__(self, product_id: int):
        self.product_id = product_id
        super().__init__(f"Product with id {product_id} was not found.")

app/db/database.py

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

from app.core.config import settings

engine = create_engine(settings.database_url, pool_pre_ping=True)

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


class Base(DeclarativeBase):
    pass


def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

app/models/product.py

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

from app.db.database import Base


class Product(Base):
    __tablename__ = "products"

    id: Mapped[int] = mapped_column(primary_key=True, index=True)
    name: Mapped[str] = mapped_column(String(100), index=True)
    description: Mapped[str] = mapped_column(String(1000))
    price: Mapped[float] = mapped_column(Numeric(10, 2))
    in_stock: Mapped[bool] = mapped_column(default=True)

app/schemas/product.py

from typing import Annotated

from pydantic import BaseModel, ConfigDict, Field


class ProductCreate(BaseModel):
    name: Annotated[str, Field(min_length=2, max_length=100)]
    description: Annotated[str, Field(min_length=1, max_length=1000)]
    price: Annotated[float, Field(gt=0)]
    in_stock: Annotated[bool, Field(default=True)]


class ProductPatch(BaseModel):
    name: Annotated[str | None, Field(default=None, min_length=2, max_length=100)]
    description: Annotated[str | None, Field(default=None, min_length=1, max_length=1000)]
    price: Annotated[float | None, Field(default=None, gt=0)]
    in_stock: Annotated[bool | None, Field(default=None)]


class ProductOut(BaseModel):
    id: int
    name: str
    description: str
    price: float
    in_stock: bool

    model_config = ConfigDict(from_attributes=True)

app/services/product_service.py

from sqlalchemy.orm import Session

from app.core.exceptions import ProductNotFoundError
from app.models.product import Product
from app.schemas.product import ProductCreate, ProductPatch


def create_product(db: Session, product_in: ProductCreate) -> Product:
    new_product = Product(**product_in.model_dump())
    db.add(new_product)
    db.commit()
    db.refresh(new_product)
    return new_product


def list_products(db: Session) -> list[Product]:
    return db.query(Product).all()


def get_product_by_id(db: Session, product_id: int) -> Product:
    product = db.get(Product, product_id)
    if product is None:
        raise ProductNotFoundError(product_id)
    return product


def patch_product(db: Session, product_id: int, data: ProductPatch) -> Product:
    product = get_product_by_id(db, product_id)

    update_data = data.model_dump(exclude_unset=True)
    for field_name, value in update_data.items():
        setattr(product, field_name, value)

    db.commit()
    db.refresh(product)
    return product


def delete_product(db: Session, product_id: int) -> None:
    product = get_product_by_id(db, product_id)
    db.delete(product)
    db.commit()

app/routers/products.py

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

from app.db.database import get_db
from app.schemas.product import ProductCreate, ProductOut, ProductPatch
from app.services.product_service import (
    create_product,
    delete_product,
    get_product_by_id,
    list_products,
    patch_product,
)

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


@router.post("/", response_model=ProductOut, status_code=status.HTTP_201_CREATED)
def create_new_product(product_in: ProductCreate, db: Session = Depends(get_db)):
    return create_product(db, product_in)


@router.get("/", response_model=list[ProductOut])
def get_all_products(db: Session = Depends(get_db)):
    return list_products(db)


@router.get("/{product_id}", response_model=ProductOut)
def get_product(product_id: int, db: Session = Depends(get_db)):
    return get_product_by_id(db, product_id)


@router.patch("/{product_id}", response_model=ProductOut)
def update_product(product_id: int, data: ProductPatch, db: Session = Depends(get_db)):
    return patch_product(db, product_id, data)


@router.delete("/{product_id}", status_code=status.HTTP_204_NO_CONTENT)
def remove_product(product_id: int, db: Session = Depends(get_db)):
    delete_product(db, product_id)
    return Response(status_code=status.HTTP_204_NO_CONTENT)

app/main.py

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

from app.core.config import settings
from app.core.exceptions import ProductNotFoundError
from app.db.database import Base, engine
from app.models import product  # noqa: F401
from app.routers import products

Base.metadata.create_all(bind=engine)

app = FastAPI(title=settings.app_name, version=settings.app_version)

app.include_router(products.router)


@app.exception_handler(ProductNotFoundError)
def product_not_found_handler(request: Request, exc: ProductNotFoundError):
    return JSONResponse(
        status_code=status.HTTP_404_NOT_FOUND,
        content={"error_code": "PRODUCT_NOT_FOUND", "detail": str(exc)},
    )


@app.get("/", tags=["Root"])
def read_root():
    return {"message": f"Welcome to {settings.app_name}"}

Detailed Code Explanation

• This program deliberately switches the example resource from User to Product, to prove the exact same layered pattern — models → schemas → services → routers — works identically for any resource, not just users.
• Product model: Numeric(10, 2) is a PostgreSQL-friendly column type for storing money accurately (10 total digits, 2 after the decimal point) — a small but important detail, since using floating-point types for currency can introduce rounding errors in real applications.
• ProductCreate requires all fields (used for POST), while ProductPatch makes every field optional (used for PATCH) — the exact same pattern established in Chapter 9.
• create_product uses Product(**product_in.model_dump()) — a shortcut that unpacks the validated Pydantic data directly into the model’s constructor, since the field names line up exactly. This avoids manually writing name=product_in.name, description=product_in.description, ... for every field.
• get_product_by_id now raises ProductNotFoundError directly (the Chapter 7 pattern) instead of returning None — and because patch_product and delete_product both call it internally, they automatically inherit correct 404 behavior for free, without repeating the check themselves.
• main.py registers a single global exception handler for ProductNotFoundError, exactly like we did for UserNotFoundError in Chapter 7 — one handler, every route covered.
• Every route in routers/products.py stays clean and readable — no manual error checks, no manual status juggling beyond what’s declared in the decorator. This is what a mature, production-shaped FastAPI router looks like once Chapters 5 through 9’s lessons are all applied together.

Execution & Testing

uv run fastapi dev app/main.py

1. Open http://127.0.0.1:8000/docs.
2. Create: POST /products/ with {"name": "Wireless Mouse", "description": "Ergonomic wireless mouse", "price": 24.99, "in_stock": true}.
3. Read all: GET /products/ — confirm your product appears.
4. Read one: GET /products/{id} — confirm you get that single product back.
5. Update: PATCH /products/{id} with just {"price": 19.99} — confirm only the price changes.
6. Delete: DELETE /products/{id} — confirm a 204 response, then GET /products/{id} again to confirm a 404 with error_code: "PRODUCT_NOT_FOUND".

You’ve now built and tested a complete, real CRUD API running on a real PostgreSQL database.

---

Tips & Instructions for Developers

• ✅ Use Docker for local databases during development — it’s fast to set up and just as fast to tear down completely, with zero leftovers on your machine.
• ✅ Always set pool_pre_ping=True when connecting to a real database server — it protects you from mysterious “connection has been closed” errors after periods of inactivity.
• ✅ Use PostgreSQL’s Numeric type for money — never plain Float, which can introduce subtle rounding errors over time.
• ✅ Keep your service layer raising domain exceptions (like ProductNotFoundError) instead of returning None — it keeps every route that depends on it automatically correct.
• ✅ Notice how little changed when moving from SQLite to PostgreSQL — treat that as validation that your project structure is doing its job. If switching databases ever requires touching your routers or schemas, something is too tightly coupled.
• ❌ Don’t hardcode real production database credentials directly in config.py — we’re doing that here only to stay focused on PostgreSQL itself. Chapter 13 fixes this properly with .env files.
• ❌ Don’t run your local development database and forget it’s running — docker ps to check, and docker stop <container_name> when you’re done for the day.

---

Official FastAPI Documentation References

• SQL (Relational) Databases
• SQLAlchemy Engine Configuration
• SQLAlchemy PostgreSQL Dialect (psycopg)
• psycopg Documentation
• Docker Official PostgreSQL Image

---

Theory Questions

1. Why is SQLite generally unsuitable for a production API deployment, even though it works fine for learning?
2. What does the dialect+driver:// prefix in a SQLAlchemy connection string actually represent?
3. Why is pool_pre_ping=True more important for PostgreSQL than it was for SQLite?
4. Why should currency values use Numeric instead of Float in a database column?
5. What does it tell you about your project’s architecture if switching databases required changes to your router or schema files?

Practical Assignment

1. Add search: Starting from Program 2’s project, add a GET /products/search?name=... endpoint that filters products by a partial, case-insensitive name match using SQLAlchemy’s .ilike().
2. Add a relationship: Starting from Program 2’s project, add a Category model with a one-to-many relationship to Product (each product belongs to one category), following the pattern from Chapter 8.
3. Container lifecycle practice: Stop your PostgreSQL Docker container (docker stop fastapi-postgres-crud) while your FastAPI app is still running, then try calling any endpoint. Observe the error you get, then restart the container (docker start fastapi-postgres-crud) and confirm the app recovers automatically on the next request, thanks to pool_pre_ping=True.

---

Summary

In this chapter, you built two complete, runnable FastAPI projects backed by real PostgreSQL databases:

• Connecting FastAPI to PostgreSQL using Docker and the psycopg driver, and confirming almost no application code needed to change from the SQLite version
• A complete CRUD API for a Product resource, combining routers, schemas, services, and global exception handling into one clean, production-shaped project