Опирается на правила: R-SQLA-MIG-1, R-SQLA-MIG-2, R-SQLA-MIG-3, R-SQLA-MIG-X1 из SQLAlchemy Style Guide → раздел 6. Миграции. Безопасность DDL-операций — по правилам PG-M-* из PostgreSQL: миграции.

Важно знать

  • Схема в production управляется только через инструмент миграций. Base.metadata.create_all() — только для тестовой изоляции, никогда не в lifespan-функции приложения.
  • Применённый changeset (Liquibase) или ревизию (Alembic) нельзя редактировать. Исправление — новая миграция поверх.
  • Liquibase — рекомендуемый инструмент для проектов, где Java-сервисы используют тот же changelog. Alembic — правомерная альтернатива для Python-only стека.
  • Сгенерированный changelog (alembic revision --autogenerate или liquibase generate-changelog) всегда требует ручного вычитывания: автогенерация не ловит смены типа, частичные индексы, sequence.
  • CREATE INDEX CONCURRENTLY не работает внутри транзакции — для таких команд Alembic требует op.execute(..., execution_options={"autocommit": True}) или Liquibase runInTransaction="false".
  • Для обратно-несовместимых изменений применяется expand-contract: сначала добавить новый столбец, переключить код, только потом удалить старый.
  • Миграции прогоняются до старта приложения: в CI — отдельным шагом, в K8s — init-контейнером или отдельным job'ом.

Схема PostgreSQL — не деталь реализации, которую приложение создаёт при запуске. Это артефакт со своей историей версий, откатами и ревью. Python-сервис не отличается в этом от Java: разница только в инструменте, не в принципе.

Liquibase в Python-проекте

R-SQLA-MIG-1 называет Liquibase основным инструментом для контрактных и многоязычных проектов. В Python-контексте Liquibase запускается как CLI или Docker-образ — он работает поверх JDBC и не требует Python-кода. Удобно: один changelog обслуживает и Java-сервис, и Python-сервис, если они разделяют схему.

db/
└── changelog/
    ├── db.changelog-master.xml
    ├── 001-create-orders.sql
    ├── 002-add-customer-index.sql
    └── 003-create-products.sql

db.changelog-master.xml:

<?xml version="1.0" encoding="UTF-8"?>
<databaseChangeLog
    xmlns="http://www.liquibase.org/xml/ns/dbchangelog"
    xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
    xsi:schemaLocation="http://www.liquibase.org/xml/ns/dbchangelog
        http://www.liquibase.org/xml/ns/dbchangelog/dbchangelog-4.20.xsd">

    <include file="001-create-orders.sql" relativeToChangelogFile="true"/>
    <include file="002-add-customer-index.sql" relativeToChangelogFile="true"/>
    <include file="003-create-products.sql" relativeToChangelogFile="true"/>
</databaseChangeLog>

Применение в CI:

liquibase \
  --url="jdbc:postgresql://localhost:5432/sber_orders" \
  --username=app_user \
  --password="${DB_PASSWORD}" \
  --changeLogFile=db/changelog/db.changelog-master.xml \
  update

Через Docker — если нет JVM в CI-образе:

docker run --rm \
  -v "$(pwd)/db/changelog:/liquibase/changelog" \
  liquibase/liquibase:4.27 \
  --url="jdbc:postgresql://host.docker.internal:5432/sber_orders" \
  --changeLogFile=/liquibase/changelog/db.changelog-master.xml \
  --username=app_user --password="${DB_PASSWORD}" \
  update

Liquibase отслеживает применённые changesets в таблице DATABASECHANGELOG. Уже применённый changeset нельзя трогать (R-SQLA-MIG-X1): Liquibase считает MD5-хэш, и несовпадение вызывает ошибку при следующем update.

Alembic для Python-only стека

Если проект не связан с Java и changelog не общий, Alembic — правомерный выбор. Он интегрируется с SQLAlchemy-моделями и поддерживает async.

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

alembic init -t async alembic

alembic/env.py — подключение к async-движку:

from logging.config import fileConfig
from sqlalchemy.ext.asyncio import async_engine_from_config
from sqlalchemy import pool
from alembic import context
from adapters.out.persistence.models import Base

config = context.config
fileConfig(config.config_file_name)

target_metadata = Base.metadata


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

    import asyncio

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

    asyncio.run(_run())


def do_run_migrations(connection):
    context.configure(connection=connection, target_metadata=target_metadata)
    with context.begin_transaction():
        context.run_migrations()


run_migrations_online()

Генерация ревизии:

alembic revision --autogenerate -m "create_orders_and_products"

Файл ревизии (всегда вычитывать вручную — R-SQLA-MIG-3):

"""create_orders_and_products

Revision ID: a3f1c9e20451
Revises:
Create Date: 2024-03-12 10:15:00.123456
"""

from alembic import op
import sqlalchemy as sa
from sqlalchemy.dialects.postgresql import UUID

revision = "a3f1c9e20451"
down_revision = None
branch_labels = None
depends_on = None


def upgrade() -> None:
    op.create_table(
        "orders",
        sa.Column("id", UUID(as_uuid=True), primary_key=True),
        sa.Column("customer_id", UUID(as_uuid=True), nullable=False),
        sa.Column("total", sa.Numeric(19, 4), nullable=False),
        sa.Column("status", sa.String(50), nullable=False),
        sa.Column(
            "created_at",
            sa.DateTime(timezone=True),
            nullable=False,
            server_default=sa.text("now()"),
        ),
    )
    op.create_table(
        "products",
        sa.Column("id", UUID(as_uuid=True), primary_key=True),
        sa.Column("sku", sa.String(100), nullable=False, unique=True),
        sa.Column("price", sa.Numeric(19, 4), nullable=False),
        sa.Column("stock", sa.Integer, nullable=False, server_default="0"),
    )


def downgrade() -> None:
    op.drop_table("products")
    op.drop_table("orders")

Применение:

alembic upgrade head

Статус:

alembic current   # текущая ревизия
alembic history   # история всех ревизий

Base.metadata.create_all() — только в тестах

R-SQLA-MIG-1 запрещает create_all() в production. Характерная ошибка — поставить его в lifespan:

# ПЛОХО
@asynccontextmanager
async def lifespan(app: FastAPI):
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    yield

Почему это плохо:

  • create_all создаёт таблицы, если их нет, но не применяет изменения к существующим. Добавленный столбец не появится в уже созданной таблице.
  • Нет истории версий, нет отката, нет ревью DDL. Состояние схемы в БД расходится с кодом при любом изменении.
  • В K8s при параллельном старте нескольких реплик create_all может гонить DDL одновременно — у PostgreSQL нет идемпотентных DDL-гарантий под конкурентной нагрузкой.

Правильная схема: миграции применяются до старта приложения, create_all — только в тестовой функции conftest.py:

# conftest.py (только для тестов)
import pytest_asyncio
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession, async_sessionmaker
from adapters.out.persistence.models import Base

@pytest_asyncio.fixture(scope="session")
async def engine():
    _engine = create_async_engine("postgresql+asyncpg://...", echo=False)
    async with _engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    yield _engine
    async with _engine.begin() as conn:
        await conn.run_sync(Base.metadata.drop_all)
    await _engine.dispose()

В production — только через alembic upgrade head в init-контейнере:

# k8s job или init-контейнер
initContainers:
  - name: db-migrate
    image: sber-orders:latest
    command: ["alembic", "upgrade", "head"]
    env:
      - name: DATABASE_URL
        valueFrom:
          secretKeyRef:
            name: db-credentials
            key: url

Безопасные DDL-операции: expand-contract и CONCURRENTLY

R-SQLA-MIG-2 отсылает к PG-M-* — правилам безопасных миграций, языко-нейтральным. Ключевые идиомы в Alembic:

Добавление NOT NULL-столбца к существующей таблице — expand-contract в трёх ревизиях:

# Ревизия 1: добавить с DEFAULT (expand)
def upgrade() -> None:
    op.add_column(
        "orders",
        sa.Column("confirmed_at", sa.DateTime(timezone=True), nullable=True),
    )

# Ревизия 2: переключить код на запись confirmed_at, дождаться деплоя
# (отдельная ревизия — отдельный коммит)

# Ревизия 3: поставить NOT NULL (contract) — только когда все строки заполнены
def upgrade() -> None:
    op.alter_column("orders", "confirmed_at", nullable=False)

Индекс CONCURRENTLY — нельзя внутри транзакции. Alembic требует явного autocommit:

def upgrade() -> None:
    op.execute(
        "CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_orders_customer_id ON orders (customer_id)",
        execution_options={"autocommit": True},
    )


def downgrade() -> None:
    op.execute(
        "DROP INDEX CONCURRENTLY IF EXISTS idx_orders_customer_id",
        execution_options={"autocommit": True},
    )

lock_timeout перед тяжёлым DDL — чтобы блокирующая миграция не держала production-трафик:

def upgrade() -> None:
    op.execute("SET lock_timeout = '2s'")
    op.add_column(
        "customers",
        sa.Column("sber_id", sa.String(64), nullable=True),
    )

Вычитывание автогенерации

R-SQLA-MIG-3: alembic revision --autogenerate сравнивает Base.metadata с состоянием БД и генерирует ревизию. Это экономит время, но не заменяет ревью:

Что не ловит автогенерация:

  • Смена типа столбца (VARCHAR(100) → TEXT) — Alembic генерирует alter_column, но не всегда верно определяет, нужен ли он.
  • Частичные индексы (WHERE status = 'ACTIVE') — не включаются в метаданные SQLAlchemy без явного объявления.
  • Sequence, материализованные представления, функции PostgreSQL — вне области видимости автогенерации.
  • server_default — Alembic сравнивает строки, и малейшее расхождение в форматировании создаёт ложное изменение.

Пример того, что нужно дописать вручную после автогенерации:

def upgrade() -> None:
    op.add_column("products", sa.Column("category", sa.String(100), nullable=True))

    op.execute(
        "CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_products_active "
        "ON products (category) WHERE status = 'ACTIVE'",
        execution_options={"autocommit": True},
    )

Ревизия после вычитывания коммитируется в репозиторий и проходит code review наравне с кодом.

Что запрещено

АнтипаттернПравилоЧто взамен
Base.metadata.create_all() в lifespan/on_startupR-SQLA-MIG-1alembic upgrade head в init-контейнере до старта app
Редактирование уже применённой ревизии или changesetR-SQLA-MIG-X1Новая ревизия поверх с исправлением
alembic revision --autogenerate без ревью результатаR-SQLA-MIG-3Вычитать файл ревизии, дополнить вручную
CREATE INDEX без CONCURRENTLY на большой таблицеPG-M-*CONCURRENTLY + execution_options={"autocommit": True}
DDL без lock_timeout в productionPG-M-*SET lock_timeout = '2s' перед ALTER
Удаление столбца одним шагом без expand-contractPG-M-*Nullable → код переключён → NOT NULL → удаление
Ручной SQL-скрипт в обход инструмента миграцийR-SQLA-MIG-1Только через ревизии Alembic или Liquibase changesets

Куда дальше

  • Repository pattern в SQLAlchemy — как репозиторий работает с таблицами, схема которых задана миграциями.
  • ORM-модели на SQLAlchemy 2.0 — DeclarativeBase, Mapped, типы для денег/времени/UUID — то, что миграция должна воспроизводить в БД.
  • PostgreSQL: миграции — язык-нейтральные правила PG-M-*: expand-contract, CONCURRENTLY, lock_timeout, нумерация changesets.