Как использовать одни и те же миграции alembic для разных баз данных

Вопрос или проблема

У меня есть рабочая конфигурация миграций alembic для моей базы данных postgres под названием my_db. Теперь я хочу использовать эти миграции в другой базе данных postgres под названием my_db_test при запуске моих юнит-тестов.

Это начало моего файла migrations/env.py (который работает совершенно нормально при обычном использовании):

import asyncio
from logging.config import fileConfig

from sqlalchemy import pool
from sqlalchemy.engine import Connection
from sqlalchemy.ext.asyncio import async_engine_from_config

from alembic import context
import os # ИЗМЕНЕН
import pkgutil  # ИЗМЕНЕН
import importlib  # ИЗМЕНЕН
from config import config as cfg  # ИЗМЕНЕН
from sqlmodel import SQLModel  # ИЗМЕНЕН

# Получите URL базы данных из конфигурации  # ИЗМЕНЕН
db_url = cfg.SQLALCHEMY_DATABASE_URI  # ИЗМЕНЕН

# Укажите путь к каталогу ваших моделей  # ИЗМЕНЕН
models_dir = os.path.join(os.path.dirname(__file__), "..", "models")  # ИЗМЕНЕН

# Динамически импортируйте все Python-файлы в каталоге моделей  # ИЗМЕНЕН
for module_info in pkgutil.iter_modules([models_dir]):  # ИЗМЕНЕН
    importlib.import_module(f"models.{module_info.name}")  # ИЗМЕНЕН

# это объект конфигурации Alembic, который предоставляет
# доступ к значениям в используемом .ini файле.
config = context.config

config.set_main_option('sqlalchemy.url', db_url)  # ИЗМЕНЕН

Идея заключалась в том, чтобы по умолчанию использовать мою обычную базу данных и переключаться на test_db, используя флаг аргумента. Я добавил аргумент парсер в этот скрипт, но, к сожалению, это не так просто, как я надеялся. Я не смог напрямую внедрить переменную в файл env.py с помощью стандартной команды, такой как: alembic upgrade head --db_url <test_db>

Поэтому я попытался создать оболочку alembic на python, которая принимает команду alembic вместе с аргументом и обновляет db_url следующим образом:

# ./alembic_wrapper.py
import sys
import argparse
from alembic import command
from alembic.config import Config
from argparse import ArgumentParser
from config import config as cfg

# Функция для парсинга аргументов командной строки
def get_db_url_from_args():
    parser = ArgumentParser(description="Переопределите URL базы данных для миграций Alembic.")
    parser.add_argument(
        "--db-url", 
        type=str, 
        help="Переопределите URL базы данных (по умолчанию: из конфигурации)"
    )
    parser.add_argument(
        "alembic_command", 
        choices=["upgrade", "downgrade", "revision", "history", "current", "stamp", "merge"],
        help="Команда Alembic для выполнения"
    )
    parser.add_argument(
        "alembic_args", 
        nargs=argparse.REMAINDER, 
        help="Аргументы для команды Alembic"
    )

    args = parser.parse_args()
    return args.db_url or cfg.SQLALCHEMY_DATABASE_URI, args.alembic_command, args.alembic_args

# Получите URL базы данных и аргументы команды Alembic
db_url, alembic_command, alembic_args = get_db_url_from_args()

# Настройте Alembic для использования соответствующего db_url
config = Config("alembic.ini")
config.set_main_option('sqlalchemy.url', db_url)
print(f"Используется URL базы данных: {db_url}")

# Выполните команду Alembic
if alembic_command == "upgrade":
    command.upgrade(config, *alembic_args)
elif alembic_command == "downgrade":
    command.downgrade(config, *alembic_args)
elif alembic_command == "revision":
    command.revision(config, *alembic_args)
elif alembic_command == "history":
    command.history(config, *alembic_args)
elif alembic_command == "current":
    command.current(config, *alembic_args)
elif alembic_command == "stamp":
    command.stamp(config, *alembic_args)
elif alembic_command == "merge":
    command.merge(config, *alembic_args)

Эта оболочка работает, если я не передаю ей аргументы. Если я просто запускаю python alembic_wrapper.py upgrade head, все работает отлично. Однако, когда я передаю db_url python alembic_wrapper.py upgrade head --db-url <db_url> (не имеет значения, test_db/prod_db или неверный URL, просто выводятся все SQL-запросы (что обычно не происходит) и фактически ничего не происходит (так как в моей базе данных нет изменений).

Вывод:

(condaenv) PS C:\Users\MyUser\Documents\visual-studio-code\my-project\backend-v2> python alembic_wrapper.py upgrade head --db-url postgresql+asyncpg://user:password@localhost:5432/my_db     
Используется URL базы данных: postgresql+asyncpg://user:password@localhost:5432/my_db    
INFO  [alembic.runtime.migration] Контекст реализует PostgresqlImpl.
INFO  [alembic.runtime.migration] Генерация статического SQL
INFO  [alembic.runtime.migration] Будет предполагаться транзакционный DDL.
BEGIN;

CREATE TABLE alembic_version (
    version_num VARCHAR(32) NOT NULL,
    CONSTRAINT alembic_version_pkc PRIMARY KEY (version_num)
);

INFO  [alembic.runtime.migration] Выполняется обновление  -> f908fe4a4afd, Первоначальная миграция, включая все таблицы
-- Выполняется обновление  -> f908fe4a4afd

-- Много SQL-запросов

COMMIT;

Я не знаю, есть ли какое-то более лучшее решение в общем, чем делать это так, но я хочу продолжать использовать только одну папку миграций для моего проекта и не хочу обновлять свой файл env.py вручную каждый раз, когда я использую test_db.

Ответ или решение

Для использования одних и тех же миграций Alembic с разными базами данных, такими как my_db и my_db_test, важно грамотно организовать настройку ваших конфигураций. Ниже представлено детальное руководство, которое поможет вам достичь этой цели, сохраняя при этом единый каталог миграций.

1. Настройка конфигурации Alembic

В вашем файле migrations/env.py необходимо динамически устанавливать URL базы данных в зависимости от переданных аргументов. Вы уже начали с написания оболочки для Alembic, что является правильным направлением.

Пример структуры:

import os
import pkgutil
import importlib
from alembic import context
from sqlalchemy.ext.asyncio import async_engine_from_config
from config import config as cfg

def set_database_url():
    # Логика для определения URL
    db_url = os.getenv("DATABASE_URL", cfg.SQLALCHEMY_DATABASE_URI)
    context.config.set_main_option('sqlalchemy.url', db_url)

# Вызов функции для установки URL
set_database_url()

# Динамический импорт всех моделей
models_dir = os.path.join(os.path.dirname(__file__), "..", "models")
for module_info in pkgutil.iter_modules([models_dir]):
    importlib.import_module(f"models.{module_info.name}")

2. Оболочка для Alembic

Ваш файл alembic_wrapper.py заключается в улучшении передачи параметра с URL базы данных. Обратите внимание, что аргументы, передаваемые в командной строке, нужно обрабатывать правильно, чтобы они корректно меняли значение DATABASE_URL.

# alembic_wrapper.py
import sys
import argparse
from alembic import command
from alembic.config import Config
from config import config as cfg

def get_db_url_from_args():
    parser = argparse.ArgumentParser(description="Change DB URL for Alembic migrations.")
    parser.add_argument("--db-url", type=str, help="Database URL to use for migration.")
    parser.add_argument("alembic_command", choices=["upgrade", "downgrade", "revision", "history", "current", "stamp", "merge"])
    parser.add_argument("alembic_args", nargs=argparse.REMAINDER)

    args = parser.parse_args()
    return args.db_url or cfg.SQLALCHEMY_DATABASE_URI, args.alembic_command, args.alembic_args

db_url, alembic_command, alembic_args = get_db_url_from_args()

config = Config("alembic.ini")
config.set_main_option('sqlalchemy.url', db_url)

# Логика выполнения Alembic
if hasattr(command, alembic_command):
    getattr(command, alembic_command)(config, *alembic_args)
else:
    print(f"Ошибка: неизвестная команда '{alembic_command}'")

3. Использование environment variables

Чтобы избежать конфликта в значениях URL, рекомендуется использовать переменные окружения (environment variables). Это позволит вам сохранять конфиденциальные данные вне кода.

export DATABASE_URL=postgresql+asyncpg://user:password@localhost:5432/my_db_test

4. Пример выполнения

После настройки файла оболочки и использования переменных окружения, вы сможете запускать миграции, просто передавая нужный URL через аргументы командной строки:

python alembic_wrapper.py upgrade head --db-url postgresql+asyncpg://user:password@localhost:5432/my_db_test

Заключение

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

Оцените материал
Добавить комментарий

Капча загружается...