Настройка alembic и скриптов миграции - nemopss/mpt-kpi GitHub Wiki

1.4.1 migrations

1.4.1.1 alembic.ini

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

# A generic, single database configuration.

[alembic]
# template used to generate migration files
# file_template = %%(rev)s_%%(slug)s

# set to 'true' to run the environment during
# the 'revision' command, regardless of autogenerate
# revision_environment = false

# Logging configuration
[loggers]
keys = root,sqlalchemy,alembic,flask_migrate

[handlers]
keys = console

[formatters]
keys = generic

[logger_root]
level = WARN
handlers = console
qualname =

[logger_sqlalchemy]
level = WARN
handlers =
qualname = sqlalchemy.engine

[logger_alembic]
level = INFO
handlers =
qualname = alembic

[logger_flask_migrate]
level = INFO
handlers =
qualname = flask_migrate

[handler_console]
class = StreamHandler
args = (sys.stderr,)
level = NOTSET
formatter = generic

[formatter_generic]
format = %(levelname)-5.5s [%(name)s] %(message)s
datefmt = %H:%M:%S

• file_template: Этот параметр задает шаблон для именования файлов миграции. По умолчанию он закомментирован;

• revision_environment: Если этот параметр установлен в true, среда будет запущена во время команды revision, независимо от автоматической генерации. По умолчанию он закомментирован.

Настройки логирования позволяют управлять уровнем и форматированием сообщений логов:

Логгеры

[loggers]
keys = root,sqlalchemy,alembic,flask_migrate

• Определяет набор логгеров, используемых в проекте. В данном случае: root, sqlalchemy, alembic и flask_migrate.

Обработчики

[handlers]
keys = console

• Определяет, какие обработчики будут использоваться для логирования. В этом случае используется только консольный обработчик.

Форматировщики

[formatters]
keys = generic

• Определяет форматировщик для сообщений логирования.

Конфигурация логгеров

[logger_root]
level = WARN
handlers = console
qualname =

[logger_sqlalchemy]
level = WARN
handlers =
qualname = sqlalchemy.engine

[logger_alembic]
level = INFO
handlers =
qualname = alembic

[logger_flask_migrate]
level = INFO
handlers =
qualname = flask_migrate

• logger_root: Корневой логгер. Установлен уровень логирования на WARN, регистрирует вывод во внешний обработчик.

• logger_sqlalchemy: Логгер для SQLAlchemy, уровень также установлен на WARN.

• logger_alembic: Логгер для Alembic, уровень INFO.

• logger_flask_migrate: Логгер для Flask-Migrate, уровень INFO.

Обработчик консоли

[handler_console]
class = StreamHandler
args = (sys.stderr,)
level = NOTSET
formatter = generic

• Это определение консольного обработчика, который выводит логи в стандартный поток ошибок (sys.stderr) с уровнем логирования NOTSET, что позволяет регистировать все уровни логов. Формат сообщений задается с помощью generic.

Формат сообщений логирования

[formatter_generic]
format = %(levelname)-5.5s [%(name)s] %(message)s
datefmt = %H:%M:%S

• Формат сообщений логов определяет стиль отображения сообщений и включает в себя уровень логирования, имя логгера и само сообщение, а также время сообщением в формате HH:MM:SS.

1.4.1.2 env.py

Файл env.py является ключевым компонентом для Alembic в проектах на базе Flask, обеспечивая настройку контекста миграций и управление подключениями к базе данных.

import logging
from logging.config import fileConfig

from flask import current_app

from alembic import context

# this is the Alembic Config object, which provides
# access to the values within the .ini file in use.
config = context.config

# Interpret the config file for Python logging.
# This line sets up loggers basically.
fileConfig(config.config_file_name)
logger = logging.getLogger('alembic.env')

def get_engine():
    try:
        # this works with Flask-SQLAlchemy<3 and Alchemical
        return current_app.extensions['migrate'].db.get_engine()
    except (TypeError, AttributeError):
        # this works with Flask-SQLAlchemy>=3
        return current_app.extensions['migrate'].db.engine

def get_engine_url():
    try:
        return get_engine().url.render_as_string(hide_password=False).replace(
            '%', '%%')
    except AttributeError:
        return str(get_engine().url).replace('%', '%%')

# add your model's MetaData object here
# for 'autogenerate' support
# from myapp import mymodel
# target_metadata = mymodel.Base.metadata
config.set_main_option('sqlalchemy.url', get_engine_url())
target_db = current_app.extensions['migrate'].db

# other values from the config, defined by the needs of env.py,
# can be acquired:
# my_important_option = config.get_main_option("my_important_option")
# ... etc.

def get_metadata():
    if hasattr(target_db, 'metadatas'):
        return target_db.metadatas[None]
    return target_db.metadata

def run_migrations_offline():
    """Run migrations in 'offline' mode.

    This configures the context with just a URL
    and not an Engine, though an Engine is acceptable
    here as well.  By skipping the Engine creation
    we don't even need a DBAPI to be available.

    Calls to context.execute() here emit the given string to the
    script output.

    """
    url = config.get_main_option("sqlalchemy.url")
    context.configure(
        url=url, target_metadata=get_metadata(), literal_binds=True
    )

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

def run_migrations_online():
    """Run migrations in 'online' mode.

    In this scenario we need to create an Engine
    and associate a connection with the context.

    """

    # this callback is used to prevent an auto-migration from being generated
    # when there are no changes to the schema
    # reference: http://alembic.zzzcomputing.com/en/latest/cookbook.html
    def process_revision_directives(context, revision, directives):
        if getattr(config.cmd_opts, 'autogenerate', False):
            script = directives[0]
            if script.upgrade_ops.is_empty():
                directives[:] = []
                logger.info('No changes in schema detected.')

    conf_args = current_app.extensions['migrate'].configure_args
    if conf_args.get("process_revision_directives") is None:
        conf_args["process_revision_directives"] = process_revision_directives

    connectable = get_engine()

    with connectable.connect() as connection:
        context.configure(
            connection=connection,
            target_metadata=get_metadata(),
            **conf_args
        )

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

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

Импортируемые модули

import logging
from logging.config import fileConfig
from flask import current_app
from alembic import context

• logging: Модуль для настройки логирования.

• flask.current_app: Позволяет получить доступ к текущему экземпляру приложения Flask.

• alembic.context: Предоставляет контекст для выполнения миграций.

Конфигурация Alembic

# this is the Alembic Config object, which provides
# access to the values within the .ini file in use.
config = context.config

• config: Представляет собой объект конфигурации Alembic, который предоставляет доступ к значениям из файла alembic.ini.

Настройка логирования

# Interpret the config file for Python logging.
# This line sets up loggers basically.
fileConfig(config.config_file_name)
logger = logging.getLogger('alembic.env')

• fileConfig: Настраивает логгеры в соответствии с конфигурацией из файла.

• logger: Создает логгер для Alembic с именем 'alembic.env'.

Получение движка базы данных

def get_engine():
    try:
        return current_app.extensions['migrate'].db.get_engine()
    except (TypeError, AttributeError):
        return current_app.extensions['migrate'].db.engine

• get_engine(): Функция возвращает движок базы данных, используя механизм расширений Flask-Migrate.

Получение URL подключения

def get_engine_url():
    try:
        return get_engine().url.render_as_string(hide_password=False).replace('%', '%%')
    except AttributeError:
        return str(get_engine().url).replace('%', '%%')

• get_engine_url(): Функция возвращает строку URL, представляющую соединение с базой данных. Она обрабатывает исключения и отображает пароль, если это необходимо.

Настройка метаданных

def get_metadata():
    if hasattr(target_db, 'metadatas'):
        return target_db.metadatas[None]
    return target_db.metadata

• get_metadata(): Функция возвращает объект метаданных, содержащий информацию о таблицах базы данных.

Оффлайн-режим

def run_migrations_offline():
    """Run migrations in 'offline' mode. 
    ...
    """
    url = config.get_main_option("sqlalchemy.url")
    context.configure(
        url=url, target_metadata=get_metadata(), literal_binds=True
    )

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

• run_migrations_offline(): Функция запускает миграции в оффлайн-режиме, настраивая контекст только с URL и метаданными.

Онлайн-режим

def run_migrations_online():
    """Run migrations in 'online' mode.
    ...
    """
    def process_revision_directives(context, revision, directives):
        ...
        
    conf_args = current_app.extensions['migrate'].configure_args
    if conf_args.get("process_revision_directives") is None:
        conf_args["process_revision_directives"] = process_revision_directives

    connectable = get_engine()

    with connectable.connect() as connection:
        context.configure(
            connection=connection,
            target_metadata=get_metadata(),
            **conf_args
        )

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

• run_migrations_online(): Функция осуществляет миграции в онлайн-режиме, создавая соединение с базой данных и настраивая контекст с необходимыми метаданными и параметрами конфигурации.

Запуск миграций

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

• В зависимости от режима работы (оффлайн или онлайн), вызывается соответствующая функция для выполнения миграций.

1.4.1.3 script.py.mako

Файл script.py.mako используется в Alembic для генерации миграционных скриптов с использованием шаблонов Mako. Этот файл содержит основные элементы, необходимые для выполнения миграций, такие как идентификаторы ревизий, функции для обновления и отката изменений базы данных.

"""${message}

Revision ID: ${up_revision}
Revises: ${down_revision | comma,n}
Create Date: ${create_date}

"""
from alembic import op
import sqlalchemy as sa
${imports if imports else ""}

# revision identifiers, used by Alembic.
revision = ${repr(up_revision)}
down_revision = ${repr(down_revision)}
branch_labels = ${repr(branch_labels)}
depends_on = ${repr(depends_on)}


def upgrade():
    ${upgrades if upgrades else "pass"}


def downgrade():
    ${downgrades if downgrades else "pass"}

Импортируемые модули

from alembic import op
import sqlalchemy as sa
${imports if imports else ""}

• from alembic import op: Импортирует объект op, используемый для выполнения операций миграции.

• import sqlalchemy as sa: Импортирует SQLAlchemy для работы с типами данных и операциями базы данных.

• imports: Здесь можно вставить дополнительные импорты, если они необходимы для миграции.

Идентификаторы ревизий

# revision identifiers, used by Alembic.
revision = ${repr(up_revision)}
down_revision = ${repr(down_revision)}
branch_labels = ${repr(branch_labels)}
depends_on = ${repr(depends_on)}

• revision: Текущий идентификатор ревизии.

• down_revision: Идентификатор предыдущей ревизии.

• branch_labels: Метки веток, относящихся к ревизии.

• depends_on: Зависимости от других ревизий, если таковые имеются.

Функция обновления

def upgrade():
    ${upgrades if upgrades else "pass"}

• upgrade(): Функция, которая выполняет изменения в базе данных для текущей ревизии. Если не заданы изменения, используется заглавный оператор pass, который ничего не делает.

Функция отката

def downgrade():
    ${downgrades if downgrades else "pass"}

• downgrade(): Функция, которая откатывает изменения миграции обратно к предыдущему состоянию. Аналогично функции upgrade, по умолчанию используется pass, если не заданы операции.

1.4.1.4 Versions

1.4.1.4.1 a32201abf005_initial_migration.py

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

"""Initial migration

Revision ID: a32201abf005
Revises: 
Create Date: 2024-12-02 06:47:03.151935

"""
from alembic import op
import sqlalchemy as sa

# revision identifiers, used by Alembic.
revision = 'a32201abf005'
down_revision = None
branch_labels = None
depends_on = None

def upgrade():
    # ### commands auto generated by Alembic - please adjust! ###
    op.create_table('position',
    sa.Column('id', sa.Integer(), nullable=False),
    sa.Column('name', sa.String(length=100), nullable=False),
    sa.Column('score_threshold', sa.Integer(), nullable=False),
    sa.PrimaryKeyConstraint('id'),
    sa.UniqueConstraint('name')
    )
    op.create_table('user',
    sa.Column('id', sa.Integer(), nullable=False),
    sa.Column('first_name', sa.String(length=50), nullable=False),
    sa.Column('last_name', sa.String(length=50), nullable=False),
    sa.Column('middle_name', sa.String(length=50), nullable=True),
    sa.Column('position_id', sa.Integer(), nullable=True),
    sa.Column('phone_number', sa.String(length=15), nullable=True),
    sa.Column('email', sa.String(length=120), nullable=False),
    sa.Column('password_hash', sa.String(length=128), nullable=True),
    sa.Column('is_admin', sa.Boolean(), nullable=True),
    sa.ForeignKeyConstraint(['position_id'], ['position.id'], ),
    sa.PrimaryKeyConstraint('id'),
    sa.UniqueConstraint('email'),
    sa.UniqueConstraint('phone_number')
    )
    # ### end Alembic commands ###

def downgrade():
    # ### commands auto generated by Alembic - please adjust! ###
    op.drop_table('user')
    op.drop_table('position')
    # ### end Alembic commands ###

Импортируемые модули

from alembic import op
import sqlalchemy as sa

• alembic.op: Модуль, предоставляющий функции для работы с операциями миграции.

• sqlalchemy: Библиотека для работы с базами данных, предоставляющая инструменты для определения и манипуляций с таблицами и их столбцами.

Идентификаторы ревизий Alembic

revision = 'a32201abf005'
down_revision = None
branch_labels = None
depends_on = None

• revision: Уникальный идентификатор для данной миграции.

• down_revision: Указывает, что данная миграция не имеет предшествующей.

• branch_labels и depends_on: Не используются в данной миграции.

Функция upgrade()

Функция upgrade() используется для применения изменений в структуре базы данных. В данном случае она включает в себя создание двух таблиц: position и user.

Создание таблицы position

op.create_table('position',
    sa.Column('id', sa.Integer(), nullable=False),
    sa.Column('name', sa.String(length=100), nullable=False),
    sa.Column('score_threshold', sa.Integer(), nullable=False),
    sa.PrimaryKeyConstraint('id'),
    sa.UniqueConstraint('name')
)

• id: Целочисленный идентификатор, является первичным ключом.

• name: Строка длиной до 100 символов, уникальная для каждой позиции.

• score_threshold: Целочисленный порог баллов для позиции.

Создание таблицы user

op.create_table('user',
    sa.Column('id', sa.Integer(), nullable=False),
    sa.Column('first_name', sa.String(length=50), nullable=False),
    sa.Column('last_name', sa.String(length=50), nullable=False),
    sa.Column('middle_name', sa.String(length=50), nullable=True),
    sa.Column('position_id', sa.Integer(), nullable=True),
    sa.Column('phone_number', sa.String(length=15), nullable=True),
    sa.Column('email', sa.String(length=120), nullable=False),
    sa.Column('password_hash', sa.String(length=128), nullable=True),
    sa.Column('is_admin', sa.Boolean(), nullable=True),
    sa.ForeignKeyConstraint(['position_id'], ['position.id'], ),
    sa.PrimaryKeyConstraint('id'),
    sa.UniqueConstraint('email'),
    sa.UniqueConstraint('phone_number')
)

• id: Целочисленный идентификатор, является первичным ключом.

• first_name: Имя пользователя, не может быть пустым.

• last_name: Фамилия пользователя, не может быть пустой.

• middle_name: Отчество пользователя (необязательный).

• position_id: Идентификатор позиции (внешний ключ, указывающий на таблицу position).

• phone_number: Номер телефона пользователя (уникальный).

• email: Адрес электронной почты (уникальный).

• password_hash: Хэш пароля пользователя (необязательный).

• is_admin: Указывает, является ли пользователь администратором (необязательный).

Функция downgrade() Функция downgrade() используется для отмены изменений, выполненных в функции upgrade(). В данном случае она удаляет созданные таблицы.

def downgrade():
    op.drop_table('user')
    op.drop_table('position')

• op.drop_table('user'): Удаляет таблицу user.

• op.drop_table('position'): Удаляет таблицу position.