Chapter 57, Backend

Alembic: database migrations

The database schema evolves along with the code. The migration is the versioned history of this evolution, applicable in any environment in a repeatable way.

The files for this chapter are in exemplos/api_pedidos/.

The problem

You add a column to the model. Your local database does not have this column, the test one neither, and the production one even less. Changing it manually in each environment is a sure mistake. The Alembic keeps each change as a versioned migration file, which knows to upgrade (upgrade) and to downgrade (downgrade) the schema, and records in a table of the database itself which version is applied.

Configuration

The alembic.ini points to the migrations folder. The env.py is the heart: it connects Alembic to your models and to the same database configuration that the application uses, so that migration and API never diverge in URL:

exemplos/api_pedidos/alembic.ini

[alembic]
script_location = %(here)s/migrations
prepend_sys_path = src
path_separator = os
file_template = %%(rev)s_%%(slug)s

[loggers]
keys = root,sqlalchemy,alembic

[handlers]
keys = console

[formatters]
keys = generic

[logger_root]
level = WARNING
handlers = console
qualname =

[logger_sqlalchemy]
level = WARNING
handlers =
qualname = sqlalchemy.engine

[logger_alembic]
level = INFO
handlers =
qualname = alembic

[handler_console]
class = StreamHandler
args = (sys.stderr,)
level = NOTSET
formatter = generic

[formatter_generic]
format = %(levelname)-5.5s [%(name)s] %(message)s
datefmt = %H:%M:%S

exemplos/api_pedidos/migrations/env.py

from logging.config import fileConfig

from alembic import context
from sqlalchemy import engine_from_config, pool

from api_pedidos.config import obter_configuracao
from api_pedidos.modelos import Base

config = context.config
if config.config_file_name is not None:
    fileConfig(config.config_file_name, disable_existing_loggers=False)

# A URL vem da configuração da aplicação, a mesma que a API usa. O "%" precisa ser escapado
# porque o ConfigParser do Alembic trata "%" como caractere especial.
url = obter_configuracao().database_url.get_secret_value()
config.set_main_option("sqlalchemy.url", url.replace("%", "%%"))

target_metadata = Base.metadata


def run_migrations_offline() -> None:
    """Gera o SQL sem conectar ao banco (alembic upgrade head --sql)."""
    context.configure(
        url=config.get_main_option("sqlalchemy.url"),
        target_metadata=target_metadata,
        literal_binds=True,
        dialect_opts={"paramstyle": "named"},
    )
    with context.begin_transaction():
        context.run_migrations()


def run_migrations_online() -> None:
    connectable = engine_from_config(
        config.get_section(config.config_ini_section, {}),
        prefix="sqlalchemy.",
        poolclass=pool.NullPool,
    )
    with connectable.connect() as conexao:
        context.configure(connection=conexao, target_metadata=target_metadata)
        with context.begin_transaction():
            context.run_migrations()


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

The target_metadata = Base.metadata is what allows the --autogenerate to compare your models with the real database.

Create a migration

The --autogenerate compares the models with the database and writes the migration for you. It only generates the draft, and review is mandatory:

Terminal

uv run alembic revision --autogenerate -m "criar pedidos" --rev-id 0001

Output

INFO  [alembic.autogenerate.compare.tables] Detected added table 'pedidos'
INFO  [alembic.autogenerate.compare.constraints] Detected added index 'ix_pedidos_cliente' on '('cliente',)'
INFO  [alembic.autogenerate.compare.tables] Detected added table 'itens_pedido'
Generating migrations/versions/0001_criar_pedidos.py ...  done

The generated file, after my review (Alembic writes comments and old typing that I clean up):

exemplos/api_pedidos/migrations/versions/0001_criar_pedidos.py

"""criar pedidos

Revision ID: 0001
Revises:
Create Date: 2026-10-06 15:46:53

"""

import sqlalchemy as sa
from alembic import op

revision: str = "0001"
down_revision: str | None = None
branch_labels: str | None = None
depends_on: str | None = None


def upgrade() -> None:
    op.create_table(
        "pedidos",
        sa.Column("id", sa.Integer(), nullable=False),
        sa.Column("cliente", sa.String(length=120), nullable=False),
        sa.Column("status", sa.String(length=20), nullable=False),
        sa.Column("chave_idempotencia", sa.String(length=80), nullable=True),
        sa.Column("criado_em", sa.DateTime(timezone=True), nullable=False),
        sa.PrimaryKeyConstraint("id"),
        sa.UniqueConstraint("chave_idempotencia"),
    )
    op.create_index("ix_pedidos_cliente", "pedidos", ["cliente"], unique=False)
    op.create_table(
        "itens_pedido",
        sa.Column("id", sa.Integer(), nullable=False),
        sa.Column("pedido_id", sa.Integer(), nullable=False),
        sa.Column("produto", sa.String(length=120), nullable=False),
        sa.Column("quantidade", sa.Integer(), nullable=False),
        sa.Column("preco_centavos", sa.Integer(), nullable=False),
        sa.ForeignKeyConstraint(["pedido_id"], ["pedidos.id"], ondelete="CASCADE"),
        sa.PrimaryKeyConstraint("id"),
    )


def downgrade() -> None:
    op.drop_table("itens_pedido")
    op.drop_index("ix_pedidos_cliente", table_name="pedidos")
    op.drop_table("pedidos")

Apply, check and undo

Terminal

export API_DATABASE_URL="postgresql+psycopg://notes:notes@localhost:5432/notes_mig"
uv run alembic upgrade head

Output

INFO  [alembic.runtime.migration] Context impl PostgresqlImpl.
INFO  [alembic.runtime.migration] Will assume transactional DDL.
INFO  [alembic.runtime.migration] Running upgrade  -> 0001, criar pedidos

PostgreSQL executes schema changes within a transaction (transactional DDL), so a migration that fails in the middle is undone entirely. This is one of the reasons why I prefer it. The commands you use all the time:

Terminal

uv run alembic current
uv run alembic check
uv run alembic downgrade base
uv run alembic upgrade head --sql

Output

0001 (head)

The check fails if the models changed without a corresponding migration (it says No new upgrade operations detected. when everything is synchronized), and the --sql prints the SQL without executing anything, useful for review by those who manage the database.

Changes that affect data

Add a column NOT NULL to a table that already has rows would fail, because the existing rows have no value. The safe pattern has three steps in the same migration: add the column accepting null, fill the existing rows, and only then make it mandatory:

Uma migração que preenche dados

def upgrade() -> None:
    op.add_column("pedidos", sa.Column("canal", sa.String(length=20), nullable=True))
    op.execute("UPDATE pedidos SET canal = 'web' WHERE canal IS NULL")
    op.alter_column("pedidos", "canal", existing_type=sa.String(length=20), nullable=False)


def downgrade() -> None:
    op.drop_column("pedidos", "canal")

To deploy without downtime the pattern is called expand and contract: first a migration that only adds (compatible with the old code), then the deployment of the new code, and only in a later migration the removal of what became obsolete. Never rename or delete a column in the same version where the code stops using it.

Rules I don't break
Never edit a migration that has already been applied in another environment: create a new one. Keep a single head (two migration lines in parallel need to be unified with alembic merge). The autogenerate does not detects column renaming: it sees a removal and a creation, which would delete the data. To rename, write op.alter_column(..., new_column_name=...) by hand.

In SQLite, altering a table is different
SQLite cannot alter existing columns. Alembic offers the batch mode (render_as_batch=True), which recreates the table. That's why I run migrations against PostgreSQL, the same database as production, and avoid having SQLite as a "test database" for migrations.