Python 17 - Changing the Database Safely: Alembic Migrations

Stop relying on create_all. Record every change to the database as a versioned, reversible step, and apply it the same way on every machine.

Where StudyDesk is now

src/studydesk/
├── __init__.py          ← lifespan runs Base.metadata.create_all()
├── models.py            ← Student, User, Job
└── ...
Dockerfile, docker-compose.yml

Tables are created by create_all at startup. By the end of this article, the schema is managed by Alembic: an alembic/ folder with a baseline migration and a second migration that adds a phone column to students, applied with alembic upgrade head on your laptop and inside Docker, and create_all is gone.

---

Part 1: The Column That Never Appeared

The office wants a phone number for each student. Aman adds a line to models.py:

phone: Mapped[str | None] = mapped_column(String(20), nullable=True)

He restarts the API. POST /students with a phone → 500:

psycopg.errors.UndefinedColumn: column "phone" of relation "students" does not exist

The model has the column. The database does not. create_all ran at startup and did nothing, because the students table already existed. create_all only creates missing tables; it never alters existing ones.

"Drop the table and let it recreate," Aman says. "On production?" asks Neha. "With three hundred real students?"

That is the problem migrations solve. A migration is a small script that changes the database from one version to the next (ALTER TABLE students ADD COLUMN phone ...) and knows how to undo itself. Migrations are kept in version control, numbered, and applied in order. Every environment — your laptop, Karan's laptop, the Docker container, production — runs the same scripts and ends up with the same schema, without losing a row.

---

Part 2: What Alembic Is

Alembic is the migration tool made by the SQLAlchemy authors. The pieces:

Piece	What it is
alembic.ini	Config file at the project root
alembic/env.py	A Python file that tells Alembic how to connect and which models to look at
alembic/versions/	One Python file per migration, each with upgrade() and downgrade()
alembic_version table	A tiny table Alembic creates in your database to remember which migration is currently applied

The workflow:

1. Change a model in models.py.
2. alembic revision --autogenerate -m "describe the change" — Alembic compares your models to the live database and writes a migration file for the difference.
3. Read the generated file. Fix anything it got wrong.
4. alembic upgrade head — apply it.
5. Commit the migration file with your code.

Everyone else pulls and runs alembic upgrade head. Done.

---

Part 3: Setting Up Alembic

Install and initialise

uv add alembic
uv run alembic init alembic

The second command creates:

studydesk/
├── alembic.ini
└── alembic/
    ├── env.py
    ├── README
    ├── script.py.mako       ← template for new migration files
    └── versions/            ← empty for now

alembic.ini

The generated file is long and mostly comments. Only one line matters, and we leave it blank on purpose, because the database URL must come from .env, not from a file in Git:

sqlalchemy.url =

Find that line (it will have a placeholder like driver://user:pass@localhost/dbname) and clear the value. Everything else can stay as generated.

alembic/env.py

This is the file to rewrite. The generated version has many commented-out options; here is the full, clean version for StudyDesk.

alembic/env.py

from logging.config import fileConfig

from alembic import context
from sqlalchemy import engine_from_config, pool

from studydesk.config import DATABASE_URL
from studydesk.database import Base
from studydesk import models  # noqa: F401  (imported so every table registers on Base)

# Alembic Config object: gives access to values in alembic.ini
config = context.config

# Set the database URL from our .env, not from alembic.ini
config.set_main_option("sqlalchemy.url", DATABASE_URL)

# Set up Python logging from alembic.ini
if config.config_file_name is not None:
    fileConfig(config.config_file_name)

# The models Alembic compares the database against
target_metadata = Base.metadata


def run_migrations_offline() -> None:
    """Generate SQL without connecting (alembic upgrade head --sql)."""
    url = config.get_main_option("sqlalchemy.url")
    context.configure(
        url=url,
        target_metadata=target_metadata,
        literal_binds=True,
        dialect_opts={"paramstyle": "named"},
        compare_type=True,
    )

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


def run_migrations_online() -> None:
    """Connect to the database and run migrations."""
    connectable = engine_from_config(
        config.get_section(config.config_ini_section, {}),
        prefix="sqlalchemy.",
        poolclass=pool.NullPool,
    )

    with connectable.connect() as connection:
        context.configure(
            connection=connection,
            target_metadata=target_metadata,
            compare_type=True,
        )

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


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

Reading it

from studydesk.config import DATABASE_URL then config.set_main_option("sqlalchemy.url", DATABASE_URL) The URL is injected from .env at run time. alembic.ini stays secret-free.

from studydesk import models # noqa: F401 Importing models runs its class definitions, which registers Student, User, and Job on Base.metadata. Without this import, Base.metadata would be empty and autogenerate would think every table should be dropped. The # noqa: F401 comment tells Ruff "yes, this import looks unused, that is intentional".

target_metadata = Base.metadata What Alembic compares against the live database.

compare_type=True Also detect column type changes (e.g. String(50) → String(100)), not only added or removed columns.

Offline vs online — generated boilerplate. Online connects and runs; offline prints the SQL instead. We use online.

Why studydesk is importable here

alembic/env.py lives at the project root, outside src/. It can still import studydesk because uv installed our package into .venv (Article 1). Always run Alembic through uv run so that venv is active.

---

Part 4: The Baseline Migration

The database already has three tables made by create_all. Alembic knows nothing about them yet. We need a baseline: a first migration that describes the current state.

Generate it

uv run alembic revision --autogenerate -m "baseline: students, users, jobs"

Alembic connects, finds that Base.metadata has three tables and the database has... the same three tables. It would normally generate an empty migration. We want a migration that creates them, so that a brand-new database (Karan's laptop, a CI runner, production) can be built from zero. Easiest way: generate against an empty database.

Create a scratch database just for this step:

psql "postgresql://postgres:postgres@localhost:5432/postgres" -c "CREATE DATABASE studydesk_scratch;"

Then run autogenerate pointing at it, by overriding the URL for one command:

DATABASE_URL=postgresql+psycopg://postgres:postgres@localhost:5432/studydesk_scratch \
  uv run alembic revision --autogenerate -m "baseline: students, users, jobs"

(Windows PowerShell: $env:DATABASE_URL="postgresql+psycopg://postgres:postgres@localhost:5432/studydesk_scratch"; uv run alembic revision --autogenerate -m "baseline: students, users, jobs" then Remove-Item Env:DATABASE_URL.)

Output:

INFO  [alembic.autogenerate.compare] Detected added table 'jobs'
INFO  [alembic.autogenerate.compare] Detected added table 'students'
INFO  [alembic.autogenerate.compare] Detected added table 'users'
  Generating .../alembic/versions/a1b2c3d4e5f6_baseline_students_users_jobs.py ...  done

Drop the scratch database; its only job was to be empty:

psql "postgresql://postgres:postgres@localhost:5432/postgres" -c "DROP DATABASE studydesk_scratch;"

Read the generated file

Your revision ID will differ; the content should match this closely.

alembic/versions/a1b2c3d4e5f6_baseline_students_users_jobs.py

"""baseline: students, users, jobs

Revision ID: a1b2c3d4e5f6
Revises:
Create Date: 2026-10-31 18:40:12.000000

"""
from typing import Sequence, Union

from alembic import op
import sqlalchemy as sa


# revision identifiers, used by Alembic.
revision: str = "a1b2c3d4e5f6"
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(
        "jobs",
        sa.Column("id", sa.Integer(), autoincrement=True, nullable=False),
        sa.Column("kind", sa.String(length=50), nullable=False),
        sa.Column("payload", sa.JSON(), nullable=False),
        sa.Column("status", sa.String(length=20), nullable=False),
        sa.Column("result", sa.JSON(), nullable=True),
        sa.Column("error", sa.Text(), nullable=True),
        sa.Column("attempts", sa.Integer(), nullable=False),
        sa.Column("max_attempts", sa.Integer(), nullable=False),
        sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.text("now()"), nullable=False),
        sa.Column("started_at", sa.DateTime(timezone=True), nullable=True),
        sa.Column("finished_at", sa.DateTime(timezone=True), nullable=True),
        sa.PrimaryKeyConstraint("id"),
    )
    op.create_index(op.f("ix_jobs_kind"), "jobs", ["kind"], unique=False)
    op.create_index(op.f("ix_jobs_status"), "jobs", ["status"], unique=False)

    op.create_table(
        "students",
        sa.Column("id", sa.Integer(), autoincrement=True, nullable=False),
        sa.Column("name", sa.String(length=50), nullable=False),
        sa.Column("email", sa.String(length=255), nullable=False),
        sa.Column("age", sa.Integer(), nullable=False),
        sa.Column("city", sa.String(length=50), nullable=False),
        sa.Column("marks", sa.Integer(), nullable=False),
        sa.PrimaryKeyConstraint("id"),
    )
    op.create_index(op.f("ix_students_email"), "students", ["email"], unique=True)

    op.create_table(
        "users",
        sa.Column("id", sa.Integer(), autoincrement=True, nullable=False),
        sa.Column("email", sa.String(length=255), nullable=False),
        sa.Column("hashed_password", sa.String(length=255), nullable=False),
        sa.Column("role", sa.String(length=20), nullable=False),
        sa.Column("is_active", sa.Boolean(), nullable=False),
        sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.text("now()"), nullable=False),
        sa.PrimaryKeyConstraint("id"),
    )
    op.create_index(op.f("ix_users_email"), "users", ["email"], unique=True)


def downgrade() -> None:
    op.drop_index(op.f("ix_users_email"), table_name="users")
    op.drop_table("users")
    op.drop_index(op.f("ix_students_email"), table_name="students")
    op.drop_table("students")
    op.drop_index(op.f("ix_jobs_status"), table_name="jobs")
    op.drop_index(op.f("ix_jobs_kind"), table_name="jobs")
    op.drop_table("jobs")

Reading a migration file

revision / down_revision Every migration has a unique ID and points to the one before it. down_revision = None means this is the first. Together they form a chain, and head is the newest link.

upgrade() Operations to move the database forward. op.create_table, op.create_index, op.add_column, and so on. op is Alembic's toolbox of schema operations.

downgrade() The exact reverse, so alembic downgrade -1 can undo this step. Autogenerate writes both.

Always read what autogenerate produced. It is reliable for adding and removing tables and columns. It cannot see Python-side default= values (only server_default), it sometimes misses renames (it sees "drop column X, add column Y"), and it cannot write data migrations ("fill the new column from the old one"). The file is a draft; you own it.

Tell the existing database it is already at this version

Your local studydesk database already has the three tables. Running upgrade would try to create them again and fail. Instead, stamp it: record in alembic_version that the baseline is already applied, without running it.

uv run alembic stamp head

INFO  [alembic.runtime.migration] Running stamp_revision  -> a1b2c3d4e5f6

Check:

uv run alembic current

a1b2c3d4e5f6 (head)

A new, empty database would instead run uv run alembic upgrade head and get all three tables from the migration. Both paths end at the same version.

---

Part 5: Removing create_all

Alembic owns the schema now. Two places still call create_all; both go.

src/studydesk/__init__.py

import logging
from contextlib import asynccontextmanager

import uvicorn
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from sqlalchemy import func, select

from .config import ALLOWED_ORIGINS, HOST, PORT, RELOAD
from .database import DbSession
from .models import Student
from .routers import auth, jobs, students

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s %(levelname)s %(name)s: %(message)s",
)


@asynccontextmanager
async def lifespan(app: FastAPI):
    # Schema is managed by Alembic (run `alembic upgrade head` before starting).
    yield


app = FastAPI(
    title="StudyDesk API",
    description="Student records API for a coaching institute.",
    version="0.5.0",
    lifespan=lifespan,
)

app.add_middleware(
    CORSMiddleware,
    allow_origins=ALLOWED_ORIGINS,
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

app.include_router(auth.router)
app.include_router(students.router)
app.include_router(jobs.router)


@app.get("/", tags=["General"])
def home():
    return {"message": "Welcome to StudyDesk API"}


@app.get("/db-check", tags=["General"], summary="Check the database connection")
def db_check(db: DbSession):
    total = db.scalar(select(func.count()).select_from(Student))
    return {"database": "ok", "students_in_db": total}


def main() -> None:
    """Start the server. Host, port and reload come from the environment."""
    uvicorn.run("studydesk:app", host=HOST, port=PORT, reload=RELOAD)

The lifespan is kept (empty) so there is an obvious place for future startup work.

src/studydesk/seed.py

import json
from pathlib import Path

from .config import ADMIN_EMAIL, ADMIN_PASSWORD
from .database import SessionLocal
from .models import Student, User
from .security import hash_password

DATA_FILE = Path(__file__).parent / "data" / "students.json"


def seed_students(db) -> None:
    if db.query(Student).count() > 0:
        print("students: table already has data, skipping.")
        return

    with open(DATA_FILE, "r", encoding="utf-8") as f:
        records = json.load(f)

    for record in records:
        record.pop("grade", None)
        db.add(Student(**record))
    db.commit()
    print(f"students: inserted {len(records)}.")


def seed_admin(db) -> None:
    if not ADMIN_EMAIL or not ADMIN_PASSWORD:
        print("admin: ADMIN_EMAIL / ADMIN_PASSWORD not set, skipping.")
        return

    existing = db.query(User).filter(User.email == ADMIN_EMAIL.lower()).first()
    if existing is not None:
        print("admin: already exists, skipping.")
        return

    db.add(
        User(
            email=ADMIN_EMAIL.lower(),
            hashed_password=hash_password(ADMIN_PASSWORD),
            role="admin",
        )
    )
    db.commit()
    print(f"admin: created {ADMIN_EMAIL}.")


def seed() -> None:
    db = SessionLocal()
    try:
        seed_students(db)
        seed_admin(db)
    finally:
        db.close()


if __name__ == "__main__":
    seed()

The seed now assumes the tables exist. Order on a fresh machine: alembic upgrade head, then python -m studydesk.seed.

---

Part 6: The Real Change — Adding phone

Now the change that started all this.

Step 1: Change the model

src/studydesk/models.py

from datetime import datetime

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

from .database import Base


class Student(Base):
    __tablename__ = "students"

    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
    name: Mapped[str] = mapped_column(String(50), nullable=False)
    email: Mapped[str] = mapped_column(String(255), unique=True, nullable=False, index=True)
    phone: Mapped[str | None] = mapped_column(String(20), nullable=True)
    age: Mapped[int] = mapped_column(Integer, nullable=False)
    city: Mapped[str] = mapped_column(String(50), nullable=False)
    marks: Mapped[int] = mapped_column(Integer, nullable=False)

    def __repr__(self) -> str:
        return f"<Student id={self.id} name={self.name!r} marks={self.marks}>"


class User(Base):
    __tablename__ = "users"

    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
    email: Mapped[str] = mapped_column(String(255), unique=True, nullable=False, index=True)
    hashed_password: Mapped[str] = mapped_column(String(255), nullable=False)
    role: Mapped[str] = mapped_column(String(20), nullable=False, default="staff")
    is_active: Mapped[bool] = mapped_column(Boolean, nullable=False, default=True)
    created_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), nullable=False, server_default=func.now()
    )

    def __repr__(self) -> str:
        return f"<User id={self.id} email={self.email!r} role={self.role}>"


class Job(Base):
    __tablename__ = "jobs"

    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
    kind: Mapped[str] = mapped_column(String(50), nullable=False, index=True)
    payload: Mapped[dict] = mapped_column(JSON, nullable=False, default=dict)
    status: Mapped[str] = mapped_column(String(20), nullable=False, default="pending", index=True)
    result: Mapped[dict | None] = mapped_column(JSON, nullable=True)
    error: Mapped[str | None] = mapped_column(Text, nullable=True)
    attempts: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
    max_attempts: Mapped[int] = mapped_column(Integer, nullable=False, default=3)
    created_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), nullable=False, server_default=func.now()
    )
    started_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
    finished_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)

    def __repr__(self) -> str:
        return f"<Job id={self.id} kind={self.kind} status={self.status}>"

nullable=True is important. The table already has rows. If the new column were NOT NULL with no default, PostgreSQL would refuse to add it ("column contains null values"). New columns on existing tables are almost always nullable or have a server_default.

Step 2: Generate the migration

uv run alembic revision --autogenerate -m "add phone to students"

INFO  [alembic.autogenerate.compare] Detected added column 'students.phone'
  Generating .../alembic/versions/b7c8d9e0f1a2_add_phone_to_students.py ...  done

Step 3: Read it

alembic/versions/b7c8d9e0f1a2_add_phone_to_students.py

"""add phone to students

Revision ID: b7c8d9e0f1a2
Revises: a1b2c3d4e5f6
Create Date: 2026-10-31 18:55:40.000000

"""
from typing import Sequence, Union

from alembic import op
import sqlalchemy as sa


# revision identifiers, used by Alembic.
revision: str = "b7c8d9e0f1a2"
down_revision: Union[str, None] = "a1b2c3d4e5f6"
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None


def upgrade() -> None:
    op.add_column("students", sa.Column("phone", sa.String(length=20), nullable=True))


def downgrade() -> None:
    op.drop_column("students", "phone")

Exactly the ALTER TABLE we wanted, and down_revision points at the baseline. The chain is now None → a1b2c3 → b7c8d9.

Step 4: Apply it

uv run alembic upgrade head

INFO  [alembic.runtime.migration] Running upgrade a1b2c3d4e5f6 -> b7c8d9e0f1a2, add phone to students

Check in psql:

\d students

The phone column is there. SELECT count(*) FROM students; → still 10. No data lost.

Step 5: Try the undo

uv run alembic downgrade -1

The column disappears. uv run alembic current shows the baseline. Then bring it back:

uv run alembic upgrade head

That round trip is the point: every change is reversible and recorded.

---

Part 7: Exposing phone Through the API

The database has the column; the API must accept and return it.

src/studydesk/schemas.py

from datetime import datetime
from typing import Literal

from pydantic import BaseModel, ConfigDict, EmailStr, Field, computed_field, field_validator


def clean_text(value: str) -> str:
    """Strip spaces, reject digits, convert to Title Case."""
    value = value.strip()
    if any(char.isdigit() for char in value):
        raise ValueError("must not contain digits")
    return value.title()


# ---------- students ----------


class StudentBase(BaseModel):
    """Fields shared by every student shape."""

    name: str = Field(min_length=2, max_length=50, description="Full name", examples=["Priya Sharma"])
    email: EmailStr = Field(description="Email address", examples=["priya@example.com"])
    phone: str | None = Field(
        default=None,
        pattern=r"^\+?[0-9]{10,15}$",
        description="Phone number, digits only, optional + prefix",
        examples=["+919876543210"],
    )
    age: int = Field(ge=5, le=100, description="Age in years", examples=[17])
    city: str = Field(min_length=2, max_length=50, description="City", examples=["Ahmedabad"])
    marks: int = Field(ge=0, le=100, description="Marks out of 100", examples=[88])

    @field_validator("name", "city")
    @classmethod
    def validate_text(cls, value: str) -> str:
        return clean_text(value)


class StudentCreate(StudentBase):
    """Input schema for POST /students."""


class StudentUpdate(BaseModel):
    """Input schema for PUT /students/{id}. Every field is optional."""

    name: str | None = Field(default=None, min_length=2, max_length=50, examples=["Priya Sharma"])
    email: EmailStr | None = Field(default=None, examples=["priya@example.com"])
    phone: str | None = Field(default=None, pattern=r"^\+?[0-9]{10,15}$", examples=["+919876543210"])
    age: int | None = Field(default=None, ge=5, le=100, examples=[17])
    city: str | None = Field(default=None, min_length=2, max_length=50, examples=["Ahmedabad"])
    marks: int | None = Field(default=None, ge=0, le=100, examples=[88])

    @field_validator("name", "city")
    @classmethod
    def validate_text(cls, value: str | None) -> str | None:
        if value is None:
            return value
        return clean_text(value)


class StudentOut(StudentBase):
    """Output schema: what every student response looks like."""

    model_config = ConfigDict(from_attributes=True)

    id: int = Field(gt=0, description="Unique student ID", examples=[1])

    @computed_field
    @property
    def grade(self) -> str:
        """Letter grade calculated from marks."""
        if self.marks >= 85:
            return "A"
        if self.marks >= 70:
            return "B"
        if self.marks >= 50:
            return "C"
        return "D"


class StudentListOut(BaseModel):
    """Output schema for GET /students."""

    total: int
    total_pages: int
    offset: int
    limit: int
    count: int
    students: list[StudentOut]


# ---------- users and auth ----------


class UserCreate(BaseModel):
    """Input schema for POST /auth/register."""

    email: EmailStr = Field(examples=["ritu@studydesk.com"])
    password: str = Field(min_length=8, max_length=128, examples=["a-strong-password"])


class LoginRequest(BaseModel):
    """Input schema for POST /auth/login."""

    email: EmailStr = Field(examples=["ritu@studydesk.com"])
    password: str = Field(min_length=1, max_length=128, examples=["a-strong-password"])


class UserOut(BaseModel):
    """Output schema for a user. Never includes the password hash."""

    model_config = ConfigDict(from_attributes=True)

    id: int
    email: EmailStr
    role: Literal["staff", "admin"]
    is_active: bool
    created_at: datetime


class Token(BaseModel):
    """Output schema for POST /auth/login."""

    access_token: str
    token_type: str = "bearer"


# ---------- jobs ----------


class JobOut(BaseModel):
    """Output schema for a background job."""

    model_config = ConfigDict(from_attributes=True)

    id: int
    kind: str
    status: Literal["pending", "running", "done", "failed"]
    payload: dict
    result: dict | None
    error: str | None
    attempts: int
    max_attempts: int
    created_at: datetime
    started_at: datetime | None
    finished_at: datetime | None

phone is optional everywhere, with a simple pattern: optional +, then 10 to 15 digits. crud.py and the router need no changes: Student(**payload.model_dump()) picks up the new field automatically, and setattr in update_student handles it.

Restart, log in, PUT /students/1 with {"phone": "+919876543210"} → 200, phone in the response. Existing students show "phone": null.

---

Part 8: Migrations in Docker

The container must run alembic upgrade head before the API starts, and the migration files must be inside the image.

Dockerfile

FROM ghcr.io/astral-sh/uv:python3.12-bookworm-slim

WORKDIR /app

ENV UV_PROJECT_ENVIRONMENT=/app/.venv \
    UV_PYTHON_DOWNLOADS=never \
    UV_COMPILE_BYTECODE=1

COPY pyproject.toml uv.lock README.md ./
RUN uv sync --frozen --no-dev --no-install-project

COPY src ./src
COPY alembic.ini ./
COPY alembic ./alembic
RUN uv sync --frozen --no-dev

ENV PATH="/app/.venv/bin:$PATH"

EXPOSE 8000

CMD ["studydesk"]

Two new COPY lines. The .dockerignore from Article 9 needs no change.

docker-compose.yml

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

  migrate:
    build: .
    container_name: studydesk-migrate
    command: ["alembic", "upgrade", "head"]
    env_file:
      - .env
    environment:
      DATABASE_URL: postgresql+psycopg://postgres:postgres@db:5432/studydesk
    depends_on:
      db:
        condition: service_healthy

  api:
    build: .
    container_name: studydesk-api
    env_file:
      - .env
    environment:
      DATABASE_URL: postgresql+psycopg://postgres:postgres@db:5432/studydesk
      HOST: 0.0.0.0
      PORT: "8000"
      RELOAD: "false"
    ports:
      - "8000:8000"
    depends_on:
      db:
        condition: service_healthy
      migrate:
        condition: service_completed_successfully

  worker:
    build: .
    container_name: studydesk-worker
    command: ["studydesk-worker"]
    env_file:
      - .env
    environment:
      DATABASE_URL: postgresql+psycopg://postgres:postgres@db:5432/studydesk
    depends_on:
      db:
        condition: service_healthy
      migrate:
        condition: service_completed_successfully

volumes:
  postgres_data:

Reading the change

migrate service A one-shot container: same image, runs alembic upgrade head, then exits. Because .env is loaded, env.py gets DATABASE_URL from the environment just as it would on your laptop (the Compose environment: override points it at db).

condition: service_completed_successfully The API and worker wait until the migration container has finished with exit code 0. If a migration fails, the API never starts with a half-migrated schema. This replaces create_all in the startup order.

If the Docker database was created in Article 9

That volume already has tables from create_all, so the first migrate run would fail trying to create them. Stamp it once, the same way as locally:

docker compose up -d db
docker compose run --rm migrate alembic stamp a1b2c3d4e5f6
docker compose up --build

(Use your real baseline revision ID.) migrate then applies only the phone migration. A fresh volume (docker compose down -v first) needs no stamp; it runs both migrations from zero.

---

Part 9: The Everyday Commands

Command	What it does
uv run alembic current	Which revision the database is at
uv run alembic history	The whole chain, oldest to newest
uv run alembic revision --autogenerate -m "msg"	Draft a migration from model changes
uv run alembic revision -m "msg"	Empty migration, for hand-written changes (data fixes, renames)
uv run alembic upgrade head	Apply everything not yet applied
uv run alembic upgrade +1 / downgrade -1	One step forward / back
uv run alembic downgrade base	Undo everything (drops all tables)
uv run alembic stamp head	Mark as applied without running

Rules the team follows

1. Never edit a migration that has been applied anywhere else. Write a new one.
2. Autogenerate, then read. Especially check downgrade() and anything involving renames or defaults.
3. New columns on existing tables are nullable or have a server_default.
4. Migration files are committed with the model change, in the same commit.
5. alembic upgrade head runs before the app starts, everywhere.

---

Where StudyDesk is after this article

studydesk/
├── alembic.ini
├── alembic/
│   ├── env.py                                        ← reads DATABASE_URL from .env, targets Base.metadata
│   ├── script.py.mako
│   └── versions/
│       ├── a1b2c3d4e5f6_baseline_students_users_jobs.py
│       └── b7c8d9e0f1a2_add_phone_to_students.py
├── Dockerfile                                        ← copies alembic/
├── docker-compose.yml                                ← migrate service
└── src/studydesk/
    ├── __init__.py                                   ← no create_all
    ├── models.py                                     ← Student.phone
    ├── schemas.py                                    ← phone in Base, Update, Out
    ├── seed.py                                       ← no create_all
    └── ...

On a fresh machine:

cp .env.example .env            # fill in values
uv sync
uv run alembic upgrade head
uv run python -m studydesk.seed
uv run studydesk

---

Concept Quiz

Answers are below. No coding needed.

Q1. Why did adding phone to models.py not add it to the database?

Q2. What is a migration, and what two functions does every migration file contain?

Q3. What is the alembic_version table for?

Q4. Why do we leave sqlalchemy.url empty in alembic.ini?

Q5. Why must env.py import studydesk.models even though it never uses the name?

Q6. What does alembic stamp head do, and when is it the right command instead of upgrade?

Q7. Why was the phone column declared nullable=True?

Q8. Name two things autogenerate cannot do reliably.

Q9. In Compose, what does condition: service_completed_successfully on the migrate dependency guarantee?

Q10. A teammate edited an already-applied migration file to fix a mistake. Why is that a problem, and what should they do instead?

---

Answers

1. create_all only creates tables that are missing. The students table already existed, so it was left untouched; it never alters existing tables.
2. A versioned script that changes the schema from one state to the next. upgrade() applies the change; downgrade() reverses it.
3. It stores the ID of the currently applied migration so Alembic knows where the database is in the chain and which migrations still need to run.
4. The URL contains a password. env.py injects it from .env at run time so no secret is committed.
5. Importing the module runs the class definitions, which register the tables on Base.metadata. Without it, target_metadata is empty and autogenerate would propose dropping every table.
6. It writes the revision ID into alembic_version without running the migration. Use it when the database already has the schema (created by create_all) and only needs to be marked as up to date.
7. The table already had rows. A NOT NULL column with no default cannot be added to a non-empty table because existing rows would violate it.
8. Detect renames (it sees drop + add), see Python-side default= values, and write data migrations that transform existing rows.
9. The API and worker start only after the migration container exits with code 0. A failed migration prevents the app from starting against a broken schema.
10. Other environments already applied the old version; they will never re-run it, so the fix never reaches them, and their databases diverge. Write a new migration that corrects the mistake.

---

Epilogue

The phone column lands on Aman's laptop, in the Docker volume, and on Karan's machine, with ten students intact in each. The change is a file in Git with a timestamp and a reason, and it can be undone with one command.

Neha looks at the growing project. "Students, users, jobs, auth, a worker, migrations. How do you know none of it broke when you added phone?" "I clicked through /docs." "All twelve routes? With the admin and the staff user? Every time?"

In the next article: pytest and API testing — a test database, fixtures for sessions and logged-in users, and a suite that checks every route in seconds.