Alembic: миграции схемы — обзор

В уроке 4 остался шрам: мы дополнили модель новым полем, create_all промолчал, а первая же вставка упала с no such column. Тогда это назвали ловушкой и пообещали вернуться с инструментом. Пришло время: сегодня — Alembic, система миграций, которая переводит схему базы от версии к версии так же аккуратно, как git переводит код от коммита к коммиту.

Сначала разберём проблему до дна на живом коде — её можно и нужно видеть своими глазами. Смысл в том, что create_all отвечает ровно за одно: создать таблицы, которых ещё нет. Всё остальное — добавить колонку, переименовать, удалить — вне его полномочий.

from sqlalchemy import create_engine, text, String
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, Session

engine = create_engine("sqlite:///:memory:")

# база версии 1: таблицу создал прошлый релиз - без priority
with engine.begin() as conn:
    conn.execute(text("CREATE TABLE tasks (id INTEGER PRIMARY KEY, title VARCHAR(80))"))

class Base(DeclarativeBase):
    pass

class Task(Base):                  # модель версии 2: поле priority уже есть
    __tablename__ = "tasks"
    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str] = mapped_column(String(80))
    priority: Mapped[int] = mapped_column(default=0)

Base.metadata.create_all(engine)   # ...и молчит: таблица уже существует

with Session(engine) as s:
    try:
        s.add(Task(title="отчёт"))     # ORM шлёт priority - а колонки нет
        s.commit()
    except Exception as e:
        print("поймали:", type(e).__name__)
        s.rollback()

поймали: OperationalError

Смотри, что произошло: create_all подошёл к таблице tasks, увидел, что она уже существует, и честно отступил — ALTER TABLE он не выполняет никогда, даже если модель радикально разошлась со схемой. Ошибка всплыла позже и в другом месте: на вставке, где ORM отправил колонку priority, которой в базе нет. Код без единого warning работает всю ночь — а падает утром, на первом же пользователе.

Миграция руками: ALTER TABLE

Лечится проблема командой ALTER: база умеет достраивать таблицу на ходу. Добавим колонку руками и убедимся, что модель и база снова совпали.

from sqlalchemy import create_engine, text, select, String, inspect
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, Session

engine = create_engine("sqlite:///:memory:")

with engine.begin() as conn:
    conn.execute(text("CREATE TABLE tasks (id INTEGER PRIMARY KEY, title VARCHAR(80))"))

# миграция руками: одна команда ALTER TABLE
with engine.begin() as conn:
    conn.execute(text("ALTER TABLE tasks ADD COLUMN priority INTEGER"))

insp = inspect(engine)
print("колонки после ALTER:", [c["name"] for c in insp.get_columns("tasks")])

class Base(DeclarativeBase):
    pass

class Task(Base):                  # модель с тем же полем
    __tablename__ = "tasks"
    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str] = mapped_column(String(80))
    priority: Mapped[int] = mapped_column(default=0)

with Session(engine) as s:
    s.add(Task(title="отчёт", priority=1))
    s.commit()
    t = s.scalars(select(Task)).first()
    print("вставка прошла:", t.title, t.priority)

колонки после ALTER: ['id', 'title', 'priority']
вставка прошла: отчёт 1

Заодно в коде появился inspect(engine) — интроспекция: инструмент, который читает фактическую схему из самой базы. Именно он в блоке напечатал список колонок, и именно на нём будет стоять весь следующий сюжет. Руками, скажем честно, мигрировать можно, пока проект один, машина одна и ты не забыл ничего. На трёх машинах и пяти разработчиках ручные ALTER превращаются в угадайку: кто выполнил, кто забыл, на какой базе какая версия схемы. Нужен учёт — журнал, где записано, какие изменения схемы уже применены.

Alembic: git для схемы

Миграции — это git для схемы базы: каждая версия знает, как перейти вперёд и как вернуться назад. Изменения схемы хранятся не в головах, а в файлах-версиях, выстроенных в цепочку: каждая ссылается на предыдущую. Alembic — официальный инструмент миграций от авторов SQLAlchemy — ведёт эту цепочку и умеет по ней двигать любую базу к текущей схеме моделей: и свежую, и полуустаревшую, и вчерашнюю.

pip install alembic

alembic init migrations
#   creating migrations/env.py ...
#   creating migrations/versions ...
#   creating alembic.ini ...

После init в проекте появляется каталог migrations с двумя главными вещами: env.py — сценарий, который Alembic запускает перед любой операцией, и versions/ — будущие файлы-версии. В env.py есть одна строка, которую обязан заполнить человек: где искать метаданные моделей.

# migrations/env.py - фрагмент
from models import Base

target_metadata = Base.metadata    # без этой строки autogenerate слеп

alembic revision --autogenerate -m "add priority"
#   Generating migrations/a1b2c3_add_priority.py ... done

Autogenerate — главная магия Alembic, хотя работает она прозаично: сравнить метаданные моделей с фактической схемой базы (по линии inspect) и записать разницу в файл версии. Файл открывают и читают глазами — это код на Python, а не бинарный лог:

# migrations/a1b2c3_add_priority.py - сгенерирован autogenerate
revision = "a1b2c3"
down_revision = None                # предыдущая версия цепочки

def upgrade():
    op.add_column("tasks", sa.Column("priority", sa.Integer(), nullable=True))

def downgrade():
    op.drop_column("tasks", "priority")

alembic upgrade head          # применить всё до последней версии
# INFO  [alembic.runtime.migration] Running upgrade -> a1b2c3

alembic current               # где мы сейчас
# a1b2c3 (head)

alembic downgrade -1          # откатить на одну версию назад
# INFO  [alembic.runtime.migration] Running downgrade a1b2c3 -> (base)

Autogenerate в миниатюре

Суть autogenerate — сравнение двух списков: колонок в метаданных модели и колонок по факту в базе. Эту разницу можно получить самому, и сейчас мы это сделаем — тот же движок, та же интроспекция, что у Alembic внутри.

from sqlalchemy import create_engine, text, String, inspect
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column

class Base(DeclarativeBase):
    pass

class Task(Base):                     # модель - новая схема
    __tablename__ = "tasks"
    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str] = mapped_column(String(80))
    priority: Mapped[int] = mapped_column(default=0)
    done: Mapped[int] = mapped_column(default=0)

engine = create_engine("sqlite:///:memory:")

with engine.begin() as conn:          # база - старая схема
    conn.execute(text("CREATE TABLE tasks (id INTEGER PRIMARY KEY, title VARCHAR(80))"))

model_cols = {c.name for c in Task.__table__.columns}
db_cols = {c["name"] for c in inspect(engine).get_columns("tasks")}
print("в модели, но нет в базе:", sorted(model_cols - db_cols))
print("в базе, но нет в модели:", sorted(db_cols - model_cols))

в модели, но нет в базе: ['done', 'priority']
в базе, но нет в модели: []

Два множества, одна операция вычитания — и вот он, план миграции: две колонки, которых база ещё не видела. Alembic делает ровно это сравнение, только ещё и записывает результат в файл с upgrade и downgrade. Когда ты в следующий раз запустишь revision --autogenerate, под капотом будет вот эта разница множеств.

Миграции станут обязательными, как только база переедет с файлового SQLite на серверную СУБД из следующего урока: там руками ALTER по всем средам уже не раскидаешь. А в финальном проекте раздела create_all останется ровно там, где ему место, — в фикстуре тестов.

create_all умеет только создавать: ALTER TABLE он не выполняет никогда. Всё, что происходит с таблицей после её рождения, — территория миграций.