Lesson 7 — migrations¶

What you will build¶

The deploy step. Until now the tables arrived because a lesson script asked for them, which is fine for a temporary database and wrong for every other one: an application cannot create its own schema at startup, cannot change a table that already holds rows, and cannot tell you what shape the database is in. A migration is all three answers at once — a versioned change, applied by a command, recorded in the database itself.

This lesson also removes the last stand-in. schema.py, which lessons 4–6 used to make their temporary databases, is deleted: the application now gets its tables the way yours will.

What you need¶

Nothing new. Alembic came with the database extra in lesson 3; this lesson is about using it.

The revision¶

A migration directory has three parts: env.py (how Alembic connects, supplied by the caller so no URL is committed), script.py.mako (the template a new revision is generated from), and versions/ — the revisions themselves, one file each, in order. The first one is the schema lesson 4's tables.py described, written out as instructions:

examples/user_management/migrations/versions/0001_create_users.py

"""create the users table

Revision ID: 0001_create_users
Revises:
Create Date: 2026-10-12

"""

from __future__ import annotations

import sqlalchemy as sa
from alembic import op

revision = "0001_create_users"
down_revision = None
branch_labels = None
depends_on = None


def upgrade() -> None:
    op.create_table(
        "users",
        sa.Column("id", sa.Integer(), primary_key=True, autoincrement=True),
        sa.Column("email", sa.String(length=320), nullable=False),
        sa.Column("display_name", sa.String(length=200), nullable=False),
        sa.Column("enabled", sa.Boolean(), nullable=False, server_default=sa.text("1")),
        sa.UniqueConstraint("email", name="uq_users_email"),
    )


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

Two attributes make the chain: revision is this file's own name, and down_revision is the one it follows. None here means it is the first. upgrade() is what runs going forwards, downgrade() what runs going back, and a revision without a downgrade is a change nobody can undo.

examples/user_management/migrate.py is the small entry point the project ships: it builds the database URL from the same SqliteSettings the application uses, and applies every revision up to a given point:

examples/user_management/migrate.py

"""Apply the application's migrations.

This is the deploy step: a command someone runs against a database, never something the application
does to itself at startup.

Usage: python examples/user_management/migrate.py [path/to/app.db]
"""

from __future__ import annotations

import sys
from pathlib import Path

from xyberos.database import SqliteSettings, run_migrations

MIGRATIONS = Path(__file__).parent / "migrations"


def migrate(settings: SqliteSettings, *, revision: str = "head") -> None:
    """Apply every revision up to ``revision`` to this database."""
    file_path = settings.file_path
    if file_path is not None:
        file_path.parent.mkdir(parents=True, exist_ok=True)
    run_migrations(MIGRATIONS, url=settings.sync_url, revision=revision)


def main(argv: list[str]) -> int:
    """Apply the migrations, for a deployment step or a local reset."""
    settings = SqliteSettings(path=argv[1] if len(argv) > 1 else "var/app.db")
    migrate(settings)
    print(f"migrated {settings.sync_url} to head")
    return 0


if __name__ == "__main__":
    raise SystemExit(main(sys.argv))

What the lesson runs¶

examples/learn_traditional/traditional07_migrations.py

"""Lesson 7 — the schema as a version, applied by a command.

Run it with:

    python -m learn_traditional.traditional07_migrations
"""

from __future__ import annotations

import contextlib
import io
import tempfile
from pathlib import Path

from sqlalchemy import create_engine, text
from user_management.app import create_app
from user_management.identity import PrincipalVerifier
from user_management.migrate import MIGRATIONS, migrate

from xyberos.cli.main import main as xyberos_main
from xyberos.database import SqliteSettings
from xyberos.http import HttpAdapter
from xyberos.http.testing import create_test_client

SECRET = "a-secret-the-deployment-keeps"
#: The administrator's assertion, presented once and carried by every request.
ADMIN_HEADERS = PrincipalVerifier(SECRET).headers("admin@idp", "admin")


def revision(settings: SqliteSettings) -> str:
    """Read the revision the database is currently at."""
    engine = create_engine(settings.sync_url)
    try:
        with engine.connect() as connection:
            recorded = connection.execute(text("SELECT version_num FROM alembic_version"))
            return str(recorded.scalar_one())
    finally:
        engine.dispose()


def columns(settings: SqliteSettings) -> list[str]:
    """Read the column names of the users table, as the database itself reports them."""
    engine = create_engine(settings.sync_url)
    try:
        with engine.connect() as connection:
            described = connection.execute(text("PRAGMA table_info(users)"))
            return [str(row[1]) for row in described]
    finally:
        engine.dispose()


def deploy_command(settings: SqliteSettings) -> str:
    """Run the documented deploy command, and report it without the temporary paths."""
    captured = io.StringIO()
    with contextlib.redirect_stdout(captured):
        code = xyberos_main(
            ["migrate", "--url", settings.sync_url, "--migrations", str(MIGRATIONS)]
        )
    report = captured.getvalue().strip()
    readable = report.replace(str(MIGRATIONS), "<migrations>").replace(
        settings.sync_url, "<database>"
    )
    return f"{readable}, exit {code}"


def main() -> None:
    with tempfile.TemporaryDirectory() as directory:
        settings = SqliteSettings(path=Path(directory) / "users.db")

        migrate(settings)
        print("migrated to", revision(settings))
        print("users columns:", columns(settings))

        app = create_app(settings=settings, secret=SECRET)
        with create_test_client(HttpAdapter(app)) as client:
            client.headers.update(ADMIN_HEADERS)
            created = client.post(
                "/users", json={"email": "ada@example.com", "display_name": "Ada Lovelace"}
            )
            print("created:", created.json())

            print("the deploy step:", deploy_command(settings))

            migrate(settings)
            print("migrating again leaves the revision at", revision(settings))
            print("the user that existed before is still there:", client.get("/users/1").json())


if __name__ == "__main__":
    main()

How to run it¶

PowerShellPOSIX

$env:PYTHONPATH = "src;examples"
python -m learn_traditional.traditional07_migrations

PYTHONPATH=src:examples python -m learn_traditional.traditional07_migrations

What it prints¶

migrated to 0002_add_external_subject
users columns: ['id', 'email', 'display_name', 'enabled', 'external_subject']
created: {'id': 1, 'email': 'ada@example.com', 'display_name': 'Ada Lovelace', 'enabled': True, 'external_subject': None}
the deploy step: migrations: applied <migrations> to <database>, exit 0
migrating again leaves the revision at 0002_add_external_subject
the user that existed before is still there: {'id': 1, 'email': 'ada@example.com', 'display_name': 'Ada Lovelace', 'enabled': True, 'external_subject': None}

Read the first two lines as the job: an empty file becomes the users table, and the database records which revision produced it. Then a user is created, and the fourth line is the deploy step itself — this lesson's script calls the command your deployment runs, with the temporary paths replaced so the output is the same on your machine:

python -m xyberos migrate --url "sqlite:///var/app.db" --migrations examples/user_management/migrations

The last two lines are why this lesson exists. Migrating again is a no-op — the revision is already applied, so nothing runs and the row created before is still there. That is what metadata.create_all could never give you: it has no idea what has already happened.

What `metadata.create_all` cannot do¶

Job	`metadata.create_all`	Migrations
Create the schema	yes, if it is missing	yes, as revision `0001`
Change a table that has rows	no — it only creates tables it does not find	yes, and the rows are carried across
Tell you the current shape	no	`alembic_version` holds the revision
Run at the right time	whenever it is called, which is what makes it convenient	as a deploy step, before the new process serves traffic

Keep metadata.create_all for a throwaway database in a test — that is what lessons 3–6 did — and keep it out of an application. The moment two environments hold different data, "create the tables if they are missing" is a run that silently does nothing while the code expects a column that is not there.

Common mistakes¶

Editing a revision that has already run somewhere. A revision is a name for a change that happened; changing it means two databases at the same revision with different shapes. Write 0002.
Leaving out downgrade. It is untested code by definition, but a rollback that does not exist is worse: a deploy that has to go back needs a path, even if you never take it.
Adding a NOT NULL column to a populated table in one step. SQLite cannot do it in place. Add it nullable with a default, backfill, then tighten — or use batch_alter_table, which rebuilds the table, as the exercise does.
Migrating at startup. Every process would try, on every boot, and a failure becomes a container that never becomes healthy. The deploy step runs it once, before the new version takes traffic.
Committing the database URL. env.py has none: the caller supplies it, which also means the same migrations run in tests, in CI, and in production.
Assuming autogenerate is a review. alembic revision --autogenerate compares tables.py with a live database and proposes a revision; the file is a draft, and the commit is yours.

Exercise¶

Add a nullable nickname column to users as the revision that follows the current head, using batch mode so the table is rebuilt. Then prove it on a database that already has a user: apply the project's revisions, then yours, and show the row keeping its id and email while gaining the new column.

Show the answer

examples/learn_traditional/migrations_exercise/0003_add_nickname.py

"""add a nickname to users, rebuilding the table in SQLite batch mode

SQLite cannot alter a column in place, so ``batch_alter_table`` rebuilds the table: it creates a new
one, copies the rows into it, and swaps it in. That is why this revision is safe to run against a
database that already holds users.

``down_revision`` is the current head — the last revision in the directory, which this file follows.
A migration that names the wrong predecessor is either a no-op or a branch, and Alembic refuses a
directory with two heads rather than guessing.

Revision ID: 0003_add_nickname
Revises: 0002_add_external_subject
Create Date: 2026-10-12

"""

from __future__ import annotations

import sqlalchemy as sa
from alembic import op

revision = "0003_add_nickname"
down_revision = "0002_add_external_subject"
branch_labels = None
depends_on = None


def upgrade() -> None:
    with op.batch_alter_table("users") as batch:
        batch.add_column(sa.Column("nickname", sa.String(length=100), nullable=True))


def downgrade() -> None:
    with op.batch_alter_table("users") as batch:
        batch.drop_column("nickname")

examples/learn_traditional/traditional07_exercise.py

"""Lesson 7, exercise — a revision that changes a table which already has rows.

The answer to the lesson's exercise. The revision is a file next to this module; the answer copies
the project's migration directory to a temporary one and drops the revision into it, which is what
you would do in your project by writing the file into ``migrations/versions/``.
"""

from __future__ import annotations

import shutil
import tempfile
from pathlib import Path
from typing import Any

from sqlalchemy import create_engine, text
from user_management.migrate import MIGRATIONS, migrate

from xyberos.database import SqliteSettings, run_migrations

EXERCISE_MIGRATIONS = Path(__file__).parent / "migrations_exercise"


def read(settings: SqliteSettings, statement: str) -> Any:
    """Run one statement against the database and return what it produced."""
    engine = create_engine(settings.sync_url)
    try:
        with engine.connect() as connection:
            return connection.execute(text(statement)).all()
    finally:
        engine.dispose()


def revision(settings: SqliteSettings) -> str:
    """Read the revision the database is currently at."""
    return str(read(settings, "SELECT version_num FROM alembic_version")[0][0])


def columns(settings: SqliteSettings) -> list[str]:
    """Read the column names of the users table."""
    return [str(row[1]) for row in read(settings, "PRAGMA table_info(users)")]


def the_row(settings: SqliteSettings) -> Any:
    """Read the only row in the table."""
    return dict(read(settings, "SELECT * FROM users")[0]._mapping)


def main() -> None:
    """Install the schema, add a row, then change the table underneath it."""
    with tempfile.TemporaryDirectory() as directory:
        settings = SqliteSettings(path=Path(directory) / "users.db")
        migrate(settings, revision="0001_create_users")

        print("migrated to", revision(settings))
        print("users columns:", columns(settings))

        engine = create_engine(settings.sync_url)
        try:
            with engine.begin() as connection:
                connection.execute(
                    text(
                        "INSERT INTO users (email, display_name, enabled) "
                        "VALUES ('ada@example.com', 'Ada Lovelace', 1)"
                    )
                )
        finally:
            engine.dispose()
        print("the row before the change:", the_row(settings))

        working = Path(directory) / "migrations"
        shutil.copytree(MIGRATIONS, working)
        shutil.copy(EXERCISE_MIGRATIONS / "0003_add_nickname.py", working / "versions")
        run_migrations(working, url=settings.sync_url)

        print("migrated to", revision(settings))
        print("users columns:", columns(settings))
        print("the same row after the change:", the_row(settings))


if __name__ == "__main__":
    main()

migrated to 0001_create_users
users columns: ['id', 'email', 'display_name', 'enabled']
the row before the change: {'id': 1, 'email': 'ada@example.com', 'display_name': 'Ada Lovelace', 'enabled': 1}
migrated to 0003_add_nickname
users columns: ['id', 'email', 'display_name', 'enabled', 'external_subject', 'nickname']
the same row after the change: {'id': 1, 'email': 'ada@example.com', 'display_name': 'Ada Lovelace', 'enabled': 1, 'external_subject': None, 'nickname': None}

The answer copies the project's migration directory to a temporary one and drops the new revision in, which is what you would do in your project by writing the file into `migrations/versions/`. Three details in the output are worth reading twice: - **The revision follows the head, not `0001`.** `down_revision` names the last revision in the directory — at this point in the path that is `0002_add_external_subject`, added by lesson 8 — and the database jumps from `0001` to `0003` in one command, applying both. A directory with two revisions claiming the same predecessor is a branch, and Alembic refuses it rather than guessing. - **`enabled` arrives as `1`, not `True`.** SQLite stores a boolean as an integer, and it is SQLAlchemy's `Boolean` type — the one `tables.py` declares — that turns it back. This answer reads `SELECT *` on purpose, to show every column the table has now, and pays for it with SQLite's own representation. - **The row keeps its `id` and `email` and gains `nickname: None`.** That is `batch_alter_table` doing the work: create a new table, copy the rows, swap it in. The alternative — dropping and recreating — is how a migration loses production data.

Next¶

Lesson 8 is the first lesson about a request that carries something the application must not trust: identity from upstream. It is also where the migration machinery starts earning its keep, because a verified principal has to be linked to a local user — and that link is a column.