Как работает Alembic?

Alembic — инструмент для миграций БД

Alembic — это легковесный инструмент управления миграциями (versioning) для SQLAlchemy и баз данных SQL. Он позволяет создавать, отслеживать и применять изменения схемы БД версионированным способом.

Основные компоненты

1. Инициализация проекта

# alembic init myproject
# Создаёт структуру:
# myproject/
#   alembic/
#     versions/          # Здесь хранятся миграции
#     env.py             # Конфиг окружения
#     script.py.mako     # Шаблон для новых миграций
#   alembic.ini          # Конфигурация Alembic

2. Конфигурация (alembic.ini)

# alembic.ini
sqlalchemy.url = postgresql://user:password@localhost/mydb

# Или через переменные окружения
# sqlalchemy.url = driver://user:pass@localhost/dbname

3. Подключение к SQLAlchemy моделям

# alembic/env.py
from logging.config import fileConfig
from sqlalchemy import engine_from_config, pool
from alembic import context
from myapp.models import Base  # Импортируем Base

config = context.config
target_metadata = Base.metadata  # Используем метаданные моделей

def run_migrations_offline():
    # Режим offline
    context.configure(url=config.get_main_option('sqlalchemy.url'))
    with context.begin_transaction():
        context.run_migrations()

def run_migrations_online():
    # Режим online
    connectable = engine_from_config(
        config.get_section(config.config_ini_section),
        prefix='sqlalchemy.',
        poolclass=pool.NullPool
    )
    with connectable.begin() as connection:
        context.configure(connection=connection, target_metadata=target_metadata)
        with context.begin_transaction():
            context.run_migrations()

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

Рабочий процесс

1. Создание миграции (автогенерация)

# alembic revision --autogenerate -m "Add user table"
# Создаёт новый файл в versions/

2. Ручное создание миграции

# alembic revision -m "Custom migration"
# Создаёт пустой шаблон

3. Структура файла миграции

# versions/001_add_users_table.py
from alembic import op
import sqlalchemy as sa

revision = '001'
down_revision = None
branch_labels = None
depends_on = None

def upgrade():
    # Команды для применения миграции (forward)
    op.create_table(
        'users',
        sa.Column('id', sa.Integer, primary_key=True),
        sa.Column('email', sa.String(255), unique=True, nullable=False),
        sa.Column('created_at', sa.DateTime, server_default=sa.func.now())
    )
    op.create_index('idx_users_email', 'users', ['email'])

def downgrade():
    # Команды для отката (backward)
    op.drop_index('idx_users_email')
    op.drop_table('users')

Команды Alembic

# Просмотр истории
# alembic history

# Текущее состояние
# alembic current

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

# Применить определённую версию
# alembic upgrade 001

# Откатить до определённой версии
# alembic downgrade 001

# Откатить последнюю миграцию
# alembic downgrade -1

# Откатить всё
# alembic downgrade base

# Переместиться вперёд на 1 версию
# alembic upgrade +1

# Переместиться назад на 1 версию
# alembic downgrade -1

Пример: Полный цикл разработки

# 1. Определяем модель
# models.py
from sqlalchemy import Column, Integer, String, DateTime
from sqlalchemy.ext.declarative import declarative_base
from datetime import datetime

Base = declarative_base()

class User(Base):
    __tablename__ = 'users'
    
    id = Column(Integer, primary_key=True)
    username = Column(String(80), unique=True, nullable=False)
    email = Column(String(120), unique=True, nullable=False)
    created_at = Column(DateTime, default=datetime.utcnow)

# 2. Генерируем миграцию
# alembic revision --autogenerate -m "Create users table"

# 3. Проверяем сгенерированный файл
# versions/002_create_users_table.py

# 4. Применяем миграцию
# alembic upgrade head

# 5. Добавляем новое поле к модели
class User(Base):
    __tablename__ = 'users'
    id = Column(Integer, primary_key=True)
    username = Column(String(80), unique=True, nullable=False)
    email = Column(String(120), unique=True, nullable=False)
    phone = Column(String(20), nullable=True)  # Новое поле
    created_at = Column(DateTime, default=datetime.utcnow)

# 6. Генерируем новую миграцию
# alembic revision --autogenerate -m "Add phone field to users"

# 7. Применяем
# alembic upgrade head

# 8. Откатываемся если что-то пошло не так
# alembic downgrade -1

Ветвление миграций

# Если работают несколько разработчиков, возможны конфликты
# Alembic поддерживает слияние веток:

revision = '003'
down_revision = ['001', '002']  # Две родительские версии

def upgrade():
    pass

def downgrade():
    pass

# Применяем слияние
# alembic upgrade head

Лучшие практики

• Создавай миграцию за каждое изменение модели
• Пиши понятные сообщения (upgrade message)
• Тестируй миграции на staging перед production
• Не редактируй старые миграции — создавай новые
• Используй downgrade для проверки отката
• Избегай данных в миграциях — разделяй на data/schema
• Используй --autogenerate с осторожностью — проверяй результат
• Версионируй миграции в git

Альтернативы Alembic

• Goose — простой инструмент на Go, работает с raw SQL
• Flyway — популярен для Java, но работает с Python
• Migrate — ещё один инструмент для SQL

Alembic — это стандартный инструмент для Python разработчиков, обеспечивающий надёжное управление изменениями схемы БД.