How can I access the "current" revision after the migration has run *inside* the env.py script?

[exhuma]

I would like to run an automation task whenever a migration task finishes successfully. At first I tried parsing the revision myself including the complexities with the -1 and head "special" revision-names.

I realised that I could potentially also use command.current to get the hash after the migration finishes. However, that command does not return a value. It relies exclusively on a side-effect.

So I tried to run alembic current in a subprocess and capture the output. This hangs. I suspect that the transaction is still active and a lock is preventing the script to finish (rough guess).

How else can I retrieve the revision hash after migrations have finished?

For context: I am dumping a snapshot of the resulting DB-schema (without data) after migration for other housekeeping tasks (integration tests). For that schema to be valid, I need to run an INSERT statement with the revision-hash that corresponds to that schema.

[zzzeek]

So I tried to run alembic current in a subprocess and capture the output. This hangs. I suspect that the transaction is still active and a lock is preventing the script to finish (rough guess).

that would not be the issue there unless you have other connections opened, the transaction scope is entirely under your control in the env.py file in your project.

alembic current works with what you are calling a "side effect" because alembic relies exclusively on the env.py file and it's transaction block to gain connectivity. however, if you have a connection already and want to use that to get "current" for the database referenced by that connection, you can use the API directly and you can get it from MigrationContext.get_current_heads(). There's an example of fetching this data at https://alembic.sqlalchemy.org/en/latest/cookbook.html#test-current-database-revision-is-at-head-s

[exhuma]

Thank you for the quick reply. This pointed me in the right direction:

def get_current_revision(connectable) -> str:
    with connectable.begin() as connection:
        context = migration.MigrationContext.configure(connection)
        return context.get_current_revision() or ""