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

[Anonymous]

У меня есть рабочая конфигурация миграции перегонного куба для моей базы данных Postgres под названием my_db.
Теперь я хочу использовать эти миграции в другой базе данных Postgres под названием my_db_test при запуске модульных тестов.< /p>
Это начало моего файла миграции/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 # EDITED
import pkgutil  # EDITED
import importlib  # EDITED
from config import config as cfg  # EDITED
from sqlmodel import SQLModel  # EDITED

# Get the database URL from the config  # EDITED
db_url = cfg.SQLALCHEMY_DATABASE_URI  # EDITED

# Specify the path to your models directory  # EDITED
models_dir = os.path.join(os.path.dirname(__file__), "..", "models")  # EDITED

# Dynamically import all Python files in the models directory  # EDITED
for module_info in pkgutil.iter_modules([models_dir]):  # EDITED
importlib.import_module(f"models.{module_info.name}")  # EDITED

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

config.set_main_option('sqlalchemy.url', db_url)  # EDITED

Идея заключалась в том, чтобы использовать мою обычную базу данных по умолчанию и переключиться на test_db с помощью флага аргумента. Я добавил в этот скрипт argpaser, но, к сожалению, это не так просто, как я надеялся.
Мне не удалось напрямую внедрить переменную в файл env.py с помощью стандартной команды, например: глава обновления перегонного куба --db_url
Поэтому я попытался создать оболочку перегонного куба в Python, которая принимает команду перегонного куба вместе с аргументом и обновляет 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

# Function to parse command-line arguments
def get_db_url_from_args():
parser = ArgumentParser(description="Override the DB URL for Alembic migrations.")
parser.add_argument(
"--db-url",
type=str,
help="Override the database URL (default: from config)"
)
parser.add_argument(
"alembic_command",
choices=["upgrade", "downgrade", "revision", "history", "current", "stamp", "merge"],
help="Alembic command to execute"
)
parser.add_argument(
"alembic_args",
nargs=argparse.REMAINDER,
help="Arguments for the Alembic command"
)

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

# Get the db URL and Alembic command arguments
db_url, alembic_command, alembic_args = get_db_url_from_args()

# Configure Alembic to use the appropriate db_url
config = Config("alembic.ini")
config.set_main_option('sqlalchemy.url', db_url)
print(f"Using database URL: {db_url}")

# Execute the Alembic command
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, он работает нормально.
Однако, когда я передаю заголовок обновления db_url python alembic_wrapper.py --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
Using database URL: postgresql+asyncpg://user:password@localhost:5432/my_db
INFO  [alembic.runtime.migration] Context impl PostgresqlImpl.
INFO  [alembic.runtime.migration] Generating static SQL
INFO  [alembic.runtime.migration] Will assume transactional DDL.
BEGIN;

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

INFO  [alembic.runtime.migration] Running upgrade  -> f908fe4a4afd, Initial migration including all tables
-- Running upgrade  -> f908fe4a4afd

-- A lot of SQL statements

COMMIT;

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

Подробнее здесь: https://stackoverflow.com/questions/791 ... -databases