Do a dry-run of an Alembic upgrade

A simple trick to allow this is to inject a conditional rollback into the run_migrations_online function in env.py that fires only when some flag is present indicating that we want a dry-run.

In case yours is already modified, recall the default implementation of the run_migrations_online function created by alembic init looks like this:

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()

Note that:

• the __enter__ method of context.begin_transaction() - which we're already calling in the default implementation - gives us a transaction object with a rollback() method, if the backend uses transactional DDL, or if transactional ddl is forced on using the transactional_ddl flag, and
• our context object has a get_x_argument method we can use to support passing custom arguments to the alembic command.

Thus, with the following small change (everything below is the same besides the addition of as transaction plus the final three lines) we can have our dry-run functionality:

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()

Now, to do a dry-run, do:

alembic -x dry-run upgrade head

and to do a real run, just do:

alembic upgrade head

like before.

You can create your own alembic config, script directory, and environment context. Critically this should have an already-created connection passed to it, which you have the ability to roll back after you're completed the dry-run.

engine = sa.create_engine(url)session = Session(bind=engine)try:    alembic_config = AlembicConfig(config_args={...})    alembic_config.attributes.update({"connection": session.connection()})    script_directory = ScriptDirectory.from_config(alembic_config)    env_context = EnvironmentContext(        alembic_config,        script_directory,        fn=lambda rev, context: script_directory._upgrade_revs("head", rev),        as_sql=False,    )    env_context.configure(connection=session.connection())    env_context.run_migrations()finally:    session.rollback()    session.close()