python - 对 Alembic 升级进行试运行

Question

有时 alembic upgrade head可能在运行时对我的生产数据库失败，即使它对我的测试数据库运行良好。例如，迁移可能会添加 NOT NULL约束以前不包含 NULL 的列s 在我的测试环境中，但确实包含 NULL s 在生产中。

在规划部署时，最好能够在运行迁移之前检查它是否能够干净利落地应用。这对于像 MySQL 这样不支持事务性 DDL(在事务中进行模式更改)的数据库来说可能是不可能的，但对于像 PostgreSQL 这样支持事务性 DDL 的数据库来说，原则上应该是可能的； Alembic 可以尝试在事务中执行升级，然后将其回滚。

(一个警告:这是一个不完美的解决方案，因为 PostgreSQL 允许某些约束为 DEFERRED ，这意味着在您提交之前不会检查它们。我想，如果不创建数据库。但是，执行 DDL 和回滚方法总比没有好。)

Alembic 是否支持此类功能？如果没有，是否有一些hacky方法来实现它？

Answer 1

允许这样做的一个简单技巧是将条件回滚注入(inject) run_migrations_online函数在 env.py仅当存在指示我们想要试运行的标志时才会触发。
如果您的已经被修改，请记忆 run_migrations_online 的默认实现由 alembic init 创建的函数看起来像这样:

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.

    """
    connectable = engine_from_config(
        config.get_section(config.config_ini_section),
        prefix="sqlalchemy.",
        poolclass=pool.NullPool,
    )

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

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

注意:

__enter__ context.begin_transaction()的方法- 我们已经在默认实现中调用了它 - 为我们提供了一个带有 rollback() 的事务对象方法，如果后端使用事务性 DDL，或者如果事务性 ddl 被强制使用 transactional_ddl 标志 , 和

我们的 context对象有一个 get_x_argument 我们可以用来支持将自定义参数传递给 alembic 的方法命令。

因此，通过以下小的更改(除了添加 as transaction 和最后三行之外，下面的所有内容都相同)我们可以拥有我们的试运行功能:

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.

    """
    connectable = engine_from_config(
        config.get_section(config.config_ini_section),
        prefix="sqlalchemy.",
        poolclass=pool.NullPool,

        # ensure the context will create a transaction
        # for backends that dont normally use transactional DDL.
        # note that ROLLBACK will not roll back DDL structures
        # on databases such as MySQL, as well as with SQLite
        # Python driver's default settings
        transactional_ddl=True,
    )

    with connectable.connect() as connection:
        context.configure(
            connection=connection, target_metadata=target_metadata
        )
        with context.begin_transaction() as transaction:
            context.run_migrations()
            if 'dry-run' in context.get_x_argument():
                print('Dry-run succeeded; now rolling back transaction...')
                transaction.rollback()

现在，要进行试运行，请执行以下操作:

alembic -x dry-run upgrade head

并进行真正的运行，只需执行以下操作:

alembic upgrade head

就像之前一样。