About Teletype
Join
Павел Логинов
@loginovpavel
Python
October 30, 2024

Хорошие практики Alembic и SQLAlchemy

В этой статье я кратко рассмотрю несколько практик, которые помогут поддерживать порядок в проекте, упростить поддержку базы данных и избежать распространённых ошибок при работе с Alembic и SQLAlchemy. Эти приёмы не раз спасали меня от проблем. Вот темы, которые мы затронем:

  1. Соглашение об именовании (Naming Convention)
  2. Сортировка миграций по дате
  3. Комментарии таблиц, колонок и миграций
  4. Работа с данными в миграциях без моделей
  5. Тестирование миграций (Stairway Test)
  6. Отдельный сервис для проведения миграций
  7. Использование миксинов для моделей

1. Соглашение об именовании (Naming Convention)

SQLAlchemy позволяет задать соглашение об именовании, которое автоматически применится ко всем таблицам и ограничениям при генерации миграций. Это избавляет от необходимости вручную задавать имена индексов, внешних ключей и других ограничений, что упрощает навигацию и делает структуру базы данных предсказуемой и унифицированной.

У меня нет рецепта как добавить конвенцию в существующий проект, кроме как перегенерировать миграции. Для нового проекта достаточно добавить соглашение у базового класса, чтобы Alembic автоматически прописывал имена в нужном формате. Пример такого формата ниже, который хорошо подходит большинству случаев:

from sqlalchemy import MetaData
from sqlalchemy.orm import DeclarativeBase

convention = {
    'all_column_names': lambda constraint, table: '_'.join(
        [column.name for column in constraint.columns.values()]
    ),
    'ix': 'ix__%(table_name)s__%(all_column_names)s',
    'uq': 'uq__%(table_name)s__%(all_column_names)s',
    'ck': 'ck__%(table_name)s__%(constraint_name)s',
    'fk': 'fk__%(table_name)s__%(all_column_names)s__%(referred_table_name)s',
    'pk': 'pk__%(table_name)s',
}

class BaseModel(DeclarativeBase):
    metadata = MetaData(naming_convention=convention)

2. Сортировка миграций по дате

По умолчанию название файла миграции в alembic начинается с тега миграции. При таком именовании, миграции в директории располагаются в случайном порядке. В некоторых случаях такой порядок неудобен. Удобнее хранить их отсортированными в хронологическом порядке.

Alembic позволяет изменять шаблон названия файла миграции. Настройка располагается в alembic.ini поле file_template. Для сортировки миграций в директории можно воспользоваться двумя удобными вариантами именования миграций.

  1. На основе даты: file_template = %%(year)d-%%(month).2d-%%(day).2d_%%(rev)s_%%(slug)s
  2. На основе времени (эпоха Unix): file_template = %%(epoch)d_%%(rev)s_%%(slug)s

Использование дат или Unix времени в именах файлов помогает держать миграции в упорядоченном виде и легко ориентироваться в них. Я предпочитаю использовать эпохи, далее в пункте 3 будет пример имен файлов.

3. Комментарии для таблиц и миграций

Следующий совет для тех, кто работает в команде. В целом хорошая практика комментировать атрибуты классов. Когда работаем с моделями SQLAlchemy, вместо комментариев в docstring, лучше воспользоваться комментариями колонок и таблиц. В этом случае комментарии будут в коде и в СУБД. Комментарии в СУБД могут помочь другим пользователям БД (ДБА или аналитикам) понимать назначение полей и таблиц.

class Event(BaseModel):
    __table_args__ = ({'comment': 'System (service) event'},)

    id: Mapped[uuid.UUID] = mapped_column(
        UUID(as_uuid=True),
        primary_key=True,
        comment='Event ID - PK',
    )
    service_id: Mapped[int] = mapped_column(
        sa.Integer,
        sa.ForeignKey(
            f'{IntegrationServiceModel.__tablename__}.id',
            ondelete='CASCADE',
        ),
        nullable=False,
        comment='FK to integration service that owns the event',
    )
    name: Mapped[str] = mapped_column(
        sa.String(256), nullable=False, comment='Event name'
    )

Также полезно комментировать миграции, чтобы легче было их находить в файловой системе. Комментарий добавляется при помощи -m <comment> при генерации миграции. Комментарий будет в докстринге и в названии файла. Согласитесь, при таком именовании найти нужную миграцию намного легче.

1728372261_c0a05e0cd317_add_integration_service.py 
1728372272_a1b4c9df789d_add_user.py
1728372283_f32d57aa1234_update_order_status.py  
1728372294_9c8e7ab45e11_create_payment.py
1728372305_bef657cd9342_remove_old_column_from_users.py

4. Не используйте модели в миграциях

Модели часто используют для манипуляций с данными, таких как перенос данных из одной таблицы в другую или изменение значений в колонках. Однако использование ORM-моделей в миграциях может привести к проблемам, если модель изменится после создания миграции. В таких случаях миграция, основанная на старой модели, сломается при её выполнении, так как схема БД может больше не соответствовать актуальной модели.

Миграции должны быть статичными и не зависеть от текущего состояния моделей, чтобы обеспечить правильное выполнение независимо от изменений в коде. Далее представлены два способа, как можно избежать использования моделей для манипуляций с данными.

Используйте сырой SQL для манипуляций с данными:

def upgrade():
    op.execute(
        "UPDATE user_account SET email = CONCAT(username, '@example.com') WHERE email IS NULL;"
    )

def downgrade():
    op.execute(
        "UPDATE user_account SET email = NULL WHERE email LIKE '%@example.com';"
    )

Определяйте таблицы непосредственно в миграции: Если вы хотите использовать SQLAlchemy для манипуляций с данными, можно определить таблицы вручную прямо в миграции. Это обеспечит статичность схемы на момент выполнения миграции и не будет зависеть от изменений в моделях.

from sqlalchemy import table, column, String

def upgrade():
    # Определяем таблицу user_account для работы с данными
    user_account = table(
        'user_account',
        column('id'),
        column('username', String),
        column('email', String)
    )

    # Получаем соединение с базой данных
    conn = op.get_bind()

    # Выбираем всех пользователей без email
    users = conn.execute(
        user_account.select().where(user_account.c.email == None)
    )

    # Обновляем email для каждого пользователя
    for user in users:
        conn.execute(
            user_account.update().where(user_account.c.id == user.id).values(email=f"{user.username}@example.com")
        )

def downgrade():
    user_account = table(
        'user_account',
        column('id'),
        column('email', String)
    )

    conn = op.get_bind()
    # Удаляем email для пользователей, добавленных в upgrade
    conn.execute(
        user_account.update().where(user_account.c.email.like('%@example.com')).values(email=None)
    )

5. Тестирование миграций лесенкой (Stairway Test)

Stairway test — это методика, при которой проверяются постепенное применение миграций лесенкой upgrade/downgrade. Это проверит миграции на то, что миграции могут создать новую БД с нуля и то, что любую миграцию можно откатить без проблем, т.е. проверяется работоспособность downgrade. Если работаете в команде, добавьте этот тест в CI, спасёт нервы и время.

Последовательность выполнения миграций при Stairway тестировании

Тест внедряется довольно легко и быстро. Оставляю ссылку на репозиторий, где можно найти пример кода. Также там есть другие полезные тесты миграций.

6. Сервис миграций

Отдельный сервис для совершения миграций. Это просто один из способов производить миграции. При разработке локально или в окружениях приближенных к разработке этот метод вписывается хорошо. Напоминаю про условный depends_on, который здесь к месту. Берём образ приложения с alembic и запускаем в отдельном контейнере. Добавляем зависимость от БД с условием, что миграции начинаются, только когда БД способен обрабатывать запросы (service_healthy). Также можно добавить условный depends_on (service_completed_successfully) на приложение, чтобы оно запускалось после успешного завершения миграций.

  db:
    image: postgres:15
    ...
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U ${POSTGRES_USER} -d ${POSTGRES_DB}"]
      interval: 10s
      start_period: 10s

  app_migrations:
    image: <app-image>
    command: [
      "python",
      "-m",
      "alembic",
      "-c",
      "<path>/alembic.ini",
      "upgrade",
      "head"
    ]
    depends_on:
      db:
        condition: service_healthy

  app:
    ...
    depends_on:
      app_migrations:
        condition: service_completed_successfully

Использование условного depends_on гарантирует, что миграции начнутся только после того, как база данных будет полностью готова к работе, а приложение запустится после завершения миграций.

7. Миксины для моделей

Да, это очевидный пункт, но всё же напомню, чтобы никто не забыл. Использование миксинов — удобный способ избежать дублирования кода. Миксины — это классы с часто используемыми полями и методами, которые можно подключить к любым моделям, где они нужны. Например, часто нам нужны поля created_at и updated_at для отслеживания времени создания и обновления записей. Также бывает полезно использовать id на основе UUID, чтобы унифицировать первичные ключи. Всё это можно вынести в миксины.

import uuid
from sqlalchemy import Column, DateTime, func
from sqlalchemy.dialects.postgresql import UUID

class TimestampMixin:
    created_at = Column(
        DateTime,
        server_default=func.now(),
        nullable=False,
        comment="Время создания записи"
    )
    updated_at = Column(
        DateTime,
        onupdate=func.now(),
        nullable=True,
        comment="Время последнего обновления записи"
    )

class UUIDPrimaryKeyMixin:
    id = Column(
        UUID(as_uuid=True),
        primary_key=True,
        default=uuid.uuid4,
        comment="Уникальный идентификатор записи"
    )

Теперь, когда у нас есть миксины, можем подключить их к любой модели, где требуется и UUID id, и метки времени:

class User(UUIDPrimaryKeyMixin, TimestampMixin, BaseModel):
    __tablename__ = 'user'
    # Другие столбцы...

Заключение

Работа с миграциями может стать настоящей головной болью, но, если соблюдать несколько простых правил, поддерживать порядок в проекте гораздо легче. Соглашения об именовании, сортировка по дате, комментарии и тестирование миграций не раз спасали нас от хаоса и помогали не совершать ошибки. Надеюсь, эта статья оказалась полезной — поделитесь своими практиками работы с миграциями в комментариях!

Павел Логинов
October 30, 2024, 06:41
1 reaction
0 views