Alembic: Основы

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

#Что такое миграции?

Миграция — это скрипт, описывающий изменения схемы базы данных:

Создание/удаление таблиц
Добавление/удаление колонок
Изменение типов данных
Создание индексов и ограничений

Зачем нужны миграции:

Версионирование схемы БД (как git для кода)
Безопасное применение изменений в production
Синхронизация схемы между разработчиками
Возможность отката (downgrade)

#Инициализация Alembic

#Установка

poetry add alembic

#Инициализация

alembic init alembic

Создаётся структура:

├── alembic/
│   ├── env.py              # Конфигурация окружения
│   ├── README
│   ├── script.py.mako      # Шаблон для миграций
│   └── versions/           # Файлы миграций
└── alembic.ini             # Главный конфиг

#Настройка alembic.ini

# alembic.ini

# URL базы данных (можно переопределить в env.py)
sqlalchemy.url = postgresql+psycopg://user:pass@localhost:5432/mydb

# Путь к моделям (для autogenerate)
# Лучше указать в env.py

#Настройка env.py

# alembic/env.py

from logging.config import fileConfig
from sqlalchemy import engine_from_config
from sqlalchemy import pool
from alembic import context

# Импортируйте ваши модели
from myapp.models import Base  # Ваш DeclarativeBase

# это объект Alembic Config
config = context.config

# Импортируйте metadata из моделей
target_metadata = Base.metadata

# Опционально: игнорировать определённые таблицы
# from alembic import op
# import sqlalchemy as sa

def include_name(name, type_, parent_name):
    """Игнорировать служебные таблицы."""
    if name == 'spatial_ref_sys':
        return False
    return True

# В run_migrations_offline() и run_migrations_online()
# добавьте include_name=include_name если нужно

#Подключение через переменные окружения

# alembic/env.py

import os
from sqlalchemy import engine_from_config
from alembic import context

def get_url():
    """Получение URL из переменных окружения."""
    return os.getenv(
        'DATABASE_URL',
        'postgresql+psycopg://postgres:postgres@localhost:5432/mydb'
    )

def run_migrations_online():
    """Run migrations in 'online' mode."""
    # Переопределяем sqlalchemy.url из env
    config.set_main_option('sqlalchemy.url', get_url())
    
    connectable = engine_from_config(
        config.get_section(config.config_ini_section, {}),
        prefix='sqlalchemy.',
        poolclass=pool.NullPool,
    )

    with connectable.connect() as connection:
        context.configure(
            connection=connection,
            target_metadata=target_metadata,
            compare_type=True,  # Сравнивать типы колонок
            compare_server_default=True,  # Сравнивать default значения
        )

        with context.begin_transaction():
            context.run_migrations()

#Создание миграций

#Авто-генерация

Alembic может автоматически создавать миграции на основе изменений в моделях:

# Сравнит metadata с БД и создаст миграцию
alembic revision --autogenerate -m "Add users table"

Что обнаруживает autogenerate:

✅ Новые таблицы
✅ Удалённые таблицы
✅ Новые колонки
✅ Удалённые колонки
✅ Изменения nullable
✅ Новые индексы
✅ Удалённые индексы
✅ Новые foreign keys

Что НЕ обнаруживает:

❌ Изменение типа колонки (иногда)
❌ Изменение имени колонки (видит как drop + add)
❌ Изменение default значения (нужен compare_server_default=True)

#Ручное создание

# Создать пустую миграцию
alembic revision -m "Add email index"

#Структура миграции

# alembic/versions/abc123456789_add_users_table.py
"""Add users table

Revision ID: abc123456789
Revises: None
Create Date: 2024-01-01 12:00:00.000000

"""
from typing import Sequence, Union

from alembic import op
import sqlalchemy as sa


# revision identifiers, used by Alembic.
revision: str = 'abc123456789'
down_revision: Union[str, None] = None  # Первая миграция
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None


def upgrade() -> None:
    """Применить изменения."""
    op.create_table(
        'users',
        sa.Column('id', sa.Integer(), nullable=False),
        sa.Column('email', sa.String(length=255), nullable=False),
        sa.Column('name', sa.String(length=100), nullable=False),
        sa.Column('created_at', sa.DateTime(), nullable=True),
        sa.PrimaryKeyConstraint('id'),
        sa.UniqueConstraint('email')
    )
    op.create_index('ix_users_email', 'users', ['email'], unique=True)


def downgrade() -> None:
    """Откатить изменения."""
    op.drop_index('ix_users_email', table_name='users')
    op.drop_table('users')

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

#Проверка статуса

# Показать текущую версию
alembic current

# Показать все миграции и статус
alembic history
alembic history --verbose

# Показать ожидающие миграции
alembic heads

#Применение

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

# Применить на конкретную ревизию
alembic upgrade abc123456789

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

# Применить на 2 вперёд
alembic upgrade +2

#Откат

# Откатить на одну миграцию
alembic downgrade -1

# Откатить к конкретной ревизии
alembic downgrade abc123456789

# Откатить все миграции (удалить все таблицы!)
alembic downgrade base

#Показать SQL без применения

# Показать SQL для миграции
alembic upgrade head --sql

# Показать SQL для отката
alembic downgrade -1 --sql

#Операции Alembic

#Таблицы

from alembic import op
import sqlalchemy as sa

def upgrade():
    # Создать таблицу
    op.create_table(
        'users',
        sa.Column('id', sa.Integer(), nullable=False),
        sa.Column('email', sa.String(255), nullable=False),
        sa.PrimaryKeyConstraint('id'),
        sa.UniqueConstraint('email')
    )
    
    # Удалить таблицу
    op.drop_table('users')
    
    # Переименовать таблицу
    op.rename_table('old_name', 'new_name')

#Колонки

def upgrade():
    # Добавить колонку
    op.add_column('users', sa.Column('age', sa.Integer(), nullable=True))
    
    # Удалить колонку
    op.drop_column('users', 'age')
    
    # Переименовать колонку
    op.alter_column('users', 'old_name', new_column_name='new_name')
    
    # Изменить тип
    op.alter_column('users', 'email',
                    existing_type=sa.String(255),
                    type_=sa.String(500))
    
    # Изменить nullable
    op.alter_column('users', 'age',
                    existing_type=sa.Integer(),
                    nullable=False)
    
    # Изменить default
    op.alter_column('users', 'created_at',
                    server_default=sa.func.now())

#Индексы

def upgrade():
    # Создать индекс
    op.create_index('ix_users_email', 'users', ['email'], unique=True)
    
    # Создать составной индекс
    op.create_index('ix_users_name_email', 'users', ['name', 'email'])
    
    # Удалить индекс
    op.drop_index('ix_users_email', table_name='users')

#Foreign Keys

def upgrade():
    # Создать foreign key
    op.create_foreign_key(
        'fk_posts_user_id',      # имя ограничения
        'posts',                  # таблица
        'users',                  # referenced таблица
        ['user_id'],              # колонки
        ['id'],                   # referenced колонки
        ondelete='CASCADE'        # опции
    )
    
    # Удалить foreign key
    op.drop_constraint('fk_posts_user_id', 'posts', type_='foreignkey')

#Unique Constraints

def upgrade():
    # Создать unique constraint
    op.create_unique_constraint('uq_users_email', 'users', ['email'])
    
    # Удалить unique constraint
    op.drop_constraint('uq_users_email', 'users', type_='unique')

#Check Constraints

def upgrade():
    # Создать check constraint
    op.create_check_constraint(
        'check_age_positive',
        'users',
        'age > 0'
    )
    
    # Удалить check constraint
    op.drop_constraint('check_age_positive', 'users', type_='check')

#Code Review миграций

#Чеклист для ревью

## Checklist для ревью миграций

### Безопасность
- [ ] downgrade() реализован и корректен
- [ ] Нет потери данных при downgrade
- [ ] Миграция идемпотентна (можно запустить дважды)

### Производительность
- [ ] Нет блокирующих операций на больших таблицах
- [ ] Индексы создаются CONCURRENTLY (для production)
- [ ] Миграция разбита на части если большая

### Качество кода
- [ ] Имена ограничений явные (не авто-генерированные)
- [ ] Типы колонок соответствуют моделям
- [ ] Nullable указан явно
- [ ] Есть комментарий к сложным изменениям

#Пример хорошей миграции

"""Add users table

Revision ID: abc123
Revises: None
Create Date: 2024-01-01 12:00:00

"""
from alembic import op
import sqlalchemy as sa

revision = 'abc123'
down_revision = None


def upgrade():
    # Создаём таблицу с явными именами ограничений
    op.create_table(
        'users',
        sa.Column('id', sa.Integer(), nullable=False),
        sa.Column('email', sa.String(length=255), nullable=False),
        sa.Column('name', sa.String(length=100), nullable=False),
        sa.Column('created_at', sa.DateTime(), server_default=sa.func.now()),
        sa.PrimaryKeyConstraint('id', name='pk_users'),
        sa.UniqueConstraint('email', name='uq_users_email')
    )
    
    # Индекс с явным именем
    op.create_index('ix_users_email', 'users', ['email'], unique=False)


def downgrade():
    # Откат в обратном порядке
    op.drop_index('ix_users_email', table_name='users')
    op.drop_table('users')

#Best Practices

#✅ Делайте

# Всегда реализуйте downgrade()
def downgrade():
    op.drop_table('users')

# Используйте явные имена ограничений
op.create_table(
    'users',
    sa.Column('id', sa.Integer(), nullable=False),
    sa.PrimaryKeyConstraint('id', name='pk_users'),
)

# Проверяйте миграции перед применением
alembic upgrade head --sql  # Посмотреть SQL

# Тестируйте downgrade
alembic upgrade head
alembic downgrade -1
alembic upgrade +1

# Используйте compare_type для autogenerate
context.configure(
    compare_type=True,
    compare_server_default=True,
)

#❌ Не делайте

# Не полагайтесь только на autogenerate
# Всегда проверяйте сгенерированную миграцию!

# Не создавайте миграции без downgrade
def upgrade():
    op.add_column('users', sa.Column('age', sa.Integer()))
# ПЛОХО: нет downgrade()

# Не используйте авто-имена ограничений
# Alembic создаст что-то вроде users_pkey
# Лучше явно: name='pk_users'

# Не применяйте миграции в production без тестирования на staging

├── alembic/ │ ├── env.py # Конфигурация окружения │ ├── README │ ├── script.py.mako # Шаблон для миграций │ └── versions/ # Файлы миграций └── alembic.ini # Главный конфиг