После autogenerate всегда:

1. Прочитай сгенерированный файл
2. Поправь если нужно (например, для rename добавь op.alter_column)
3. Протестируй на копии данных перед prod

Применение миграций

# Применить все pending миграции
alembic upgrade head

# Применить конкретную миграцию (по revision)
alembic upgrade abc123

# Применить relative (одна вперёд)
alembic upgrade +1

# Откат на одну
alembic downgrade -1

# Откат всех
alembic downgrade base

# Откат до конкретной
alembic downgrade abc123

# Текущая ревизия
alembic current

# История
alembic history --verbose

Основные операции

# Таблицы
op.create_table("users", sa.Column(...), ...)
op.drop_table("users")
op.rename_table("old_name", "new_name")

# Колонки
op.add_column("users", sa.Column("age", sa.Integer))
op.drop_column("users", "age")
op.alter_column("users", "name", new_column_name="full_name")
op.alter_column("users", "email", type_=sa.String(500))   # change type

# Индексы
op.create_index("ix_users_email", "users", ["email"], unique=True)
op.drop_index("ix_users_email", table_name="users")

# Constraints
op.create_foreign_key("fk_orders_user", "orders", "users", ["user_id"], ["id"])
op.drop_constraint("fk_orders_user", "orders", type_="foreignkey")

# Raw SQL
op.execute("UPDATE users SET active = true WHERE created_at < '2026-01-01'")

Data migrations

Иногда нужно мигрировать данные, не схему:

def upgrade():
    # Schema change
    op.add_column("users", sa.Column("full_name", sa.String(200)))

    # Data migration
    from sqlalchemy.sql import table, column
    users_table = table("users",
        column("id", sa.Integer),
        column("first_name", sa.String),
        column("last_name", sa.String),
        column("full_name", sa.String),
    )

    connection = op.get_bind()
    result = connection.execute(users_table.select())
    for row in result:
        full = f"{row.first_name} {row.last_name}"
        connection.execute(
            users_table.update().where(users_table.c.id == row.id).values(full_name=full)
        )

    # Drop old columns
    op.drop_column("users", "first_name")
    op.drop_column("users", "last_name")

Для больших таблиц - batch processing. Один UPDATE на миллион строк может зависнуть.

Multiple environments

В env.py использовать env var для URL:

config.set_main_option("sqlalchemy.url", os.getenv("DATABASE_URL"))

Запуск на разных окружениях:

# dev
DATABASE_URL=postgresql://localhost/myapp_dev alembic upgrade head

# staging
DATABASE_URL=postgresql://stage.com/myapp_stage alembic upgrade head

# prod
DATABASE_URL=postgresql://prod.com/myapp_prod alembic upgrade head

В Docker/Kubernetes URL прокидывается через env. Лучше запускать миграции в init container или CI/CD pipeline перед deployment.

Branching и merge

Если в feature branch создана миграция, и в main тоже, при merge будут две миграции с одинаковым down_revision:

alembic heads   # покажет несколько heads
alembic merge -m "merge branches" head1 head2

Создаёт merge миграцию, объединяющую branches. Это редко, но возможно при параллельной разработке.

Stamp - mark без выполнения

alembic stamp head

Помечает БД как up-to-date без выполнения миграций. Полезно когда:

• БД создана вручную (через create_all)
• Импортируешь existing DB в Alembic
• После manual fix хочешь синхронизировать metadata

Async support

С Alembic 1.6+ поддерживает async:

# env.py
from sqlalchemy.ext.asyncio import async_engine_from_config

def run_migrations_online():
    connectable = async_engine_from_config(
        config.get_section(config.config_ini_section),
        prefix="sqlalchemy.",
        poolclass=pool.NullPool,
    )

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

    asyncio.run(run())

Это сложнее sync setup. Если приложение FastAPI с async SQLAlchemy, всё равно лучше держать миграции в sync mode для простоты - они одноразовые операции.

CI/CD integration

# .github/workflows/deploy.yml
jobs:
  migrate:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with:
          python-version: "3.12"
      - run: pip install -e .
      - name: Apply migrations
        env:
          DATABASE_URL: ${{ secrets.PROD_DATABASE_URL }}
        run: alembic upgrade head

Перед каждым deployment - применить новые миграции. Если миграция fails - блокируется release.

Backward compatible migrations

Для zero-downtime deployments миграции должны быть backward compatible:

1. Добавление колонки (safe):

op.add_column("users", sa.Column("age", sa.Integer, nullable=True))

Старый код не упадёт - он не знает про новую колонку, она nullable.

2. Удаление колонки (опасно):

op.drop_column("users", "old_field")

Старый код может всё ещё писать в old_field - 500 ошибки. Стратегия:

1. Релиз 1: код перестаёт писать в old_field
2. Wait until все instances updated
3. Релиз 2: миграция удаляет колонку

3. Изменение типа (опасно):

Аналогично. Лучше: добавь новую колонку, мигрируй данные, удали старую в следующем релизе.

Распространённые ошибки

1. Не writing downgrade

def downgrade():
    pass   # ОПАСНО - нельзя откатить

Всегда implement downgrade. Когда что-то сломалось в production, нужна возможность быстро откатиться.

2. Запускать миграции прямо в app

# В app startup
Base.metadata.create_all(engine)   # БЕЗ Alembic

create_all не handles changes. И не version controlled. Используй Alembic с самого начала.

3. Изменять existing миграции

# Уже применённая миграция
def upgrade():
    op.create_table(...)
    op.create_table(...)   # добавил после deployment

Если миграция уже применена в production - изменения не выполнятся. Создавай новую миграцию.

4. Большие миграции с downtime

def upgrade():
    op.execute("UPDATE users SET ...")   # на миллионах строк - часы downtime

Batch processing, индексы concurrently, schema changes online (depending on DB). Для PostgreSQL - CONCURRENTLY для индексов, pg_repack для serious refactoring.

5. Auto-generate без проверки

Уже обсуждали. Always review autogenerate output.

Тестирование миграций

# tests/test_migrations.py
import pytest
from alembic.config import Config
from alembic import command

def test_migrations_apply():
    config = Config("alembic.ini")
    config.set_main_option("sqlalchemy.url", "sqlite:///:memory:")
    command.upgrade(config, "head")
    # Если падает - что-то не так

def test_migrations_downgrade():
    config = Config("alembic.ini")
    config.set_main_option("sqlalchemy.url", "sqlite:///:memory:")
    command.upgrade(config, "head")
    command.downgrade(config, "base")   # откат всех
    command.upgrade(config, "head")     # снова применить

Полезно для CI - убедиться что миграции и upgrade и downgrade работают.

Сравнение с Go и PHP

В Go популярен golang-migrate - простой CLI с plain SQL файлами для up/down миграций. Меньше abstraction чем Alembic, больше manual control. Работает с любой БД через одинаковый интерфейс.

В PHP стандарт - Phinx (часть Laminas/Symfony экосистемы) или встроенные миграции Symfony/Laravel. Концептуально похожи на Alembic: code-based миграции с up/down, autogenerate, version tracking.

В Python для SQLAlchemy - Alembic стандарт. Для Django - встроенные миграции (другой механизм с auto-detection).

Мини-задание

1. Setup:

mkdir myproject && cd myproject
pip install sqlalchemy alembic
alembic init alembic

# Edit alembic.ini:
# sqlalchemy.url = sqlite:///app.db

# В env.py:
# from myapp.db import Base
# target_metadata = Base.metadata

2. Create migration:

# myapp/db.py
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column

class Base(DeclarativeBase): pass

class User(Base):
    __tablename__ = "users"
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str]

alembic revision --autogenerate -m "initial schema"
alembic upgrade head
# Создаст users таблицу

3. Изменение схемы:

# Добавляем поле email
class User(Base):
    __tablename__ = "users"
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str]
    email: Mapped[str] = mapped_column(unique=True)   # новое поле

alembic revision --autogenerate -m "add email"
# Проверь сгенерированный файл!
alembic upgrade head

Что дальше

Освоили миграции с Alembic. В следующем уроке - аутентификация через JWT: реализация secure user auth для REST API.