Getting Alembic Database Version Programmatically

Question

I'm trying to figure out how to get the version of my database using Alembic. I've already gotten the database set up to use alembic and successfully performed both an upgrade and downgrade on it. I now want to get this version from my own python script.

I attempted to create a function for doing this

def get_current_database_version():
    path = os.path.join(os.path.dirname(__file__), os.path.pardir)
    alembic_cfg = Config(os.path.join(path, 'alembic.ini'))
    current_rev = command.current(alembic_cfg, head_only=True)
    return current_rev

This function returned a NoSectionError: No section: 'formatters'

I then went to my alembic.ini file to check that it had a formatters area. This is my alembic.ini file:

# A generic, single database configuration.

[alembic]
# path to migration scripts
script_location = alembic
pyramid_config_file = ../../development.ini

# template used to generate migration files
# file_template = %%(rev)s_%%(slug)s

# max length of characters to apply to the
# "slug" field
#truncate_slug_length = 40

# set to 'true' to run the environment during
# the 'revision' command, regardless of autogenerate
# revision_environment = false

# set to 'true' to allow .pyc and .pyo files without
# a source .py file to be detected as revisions in the
# versions/ directory
# sourceless = false

sqlalchemy.url = sqlite:///%(here)s/mgo.sqlite


# Logging configuration
[loggers]
keys = root,sqlalchemy,alembic

[handlers]
keys = console

[formatters]
keys = generic

[logger_root]
level = WARN
handlers = console
qualname =

[logger_sqlalchemy]
level = WARN
handlers =
qualname = sqlalchemy.engine

[logger_alembic]
level = INFO
handlers =
qualname = alembic

[handler_console]
class = StreamHandler
args = (sys.stderr,)
level = NOTSET
formatter = generic

[formatter_generic]
format = %(levelname)-5.5s [%(name)s] %(message)s
datefmt = %H:%M:%S

Anyone know what I'm doing wrong? Thanks

Edit:

Here is my attempt to use MigrationContext to solve the problem:

def get_database_revision():
    engine = create_engine("sqlite:///../mgo.db")
    conn = engine.connect()
    context = MigrationContext.configure(conn)
    current_rev = context.get_current_revision()
    return current_rev

It connects but returns none. Using sqlite browser I can see that the version in the database is not set to none.

Answer 1

You can use MigrationContext to get the current version:

from alembic.migration import MigrationContext
from sqlalchemy import create_engine

engine = create_engine("postgresql://mydatabase")
conn = engine.connect()

context = MigrationContext.configure(conn)
current_rev = context.get_current_revision()

Inside env.py you can use:

from alembic import context
migration_context = context.get_context()
current_rev = context.get_current_revision()

Lastly, it basically comes down to connecting to the database and looking at the alembic_version table. It contains the migration version as a value and that's where the database currently is (according to alembic). So you can write the code any way you want as long as that's ultimately what you're doing.

Answer 2

I suggest using stdout Config() object parameter (see here) to allow redirecting sys.stdout into a StringIO buffer as done here:

output_buffer = io.StringIO()
alembic_cfg = alembic_Config('/path/to/alembic.ini', stdout=output_buffer)
alembic_command.current(alembic_cfg)
output = output_buffer.getvalue()
print(output)

Answer 3

This question is old, but I have a solution which I think is a little simpler than those given so far.

The main observation is that when command.current is called, alembic doesn't use Python's built-in print function but instead uses the print_stdout method on the config object. Therefore, to capture the output just overload the print_stdout function! This works for me:

def get_current_database_version():
    path = os.path.join(os.path.dirname(__file__), os.path.pardir)
    alembic_cfg = Config(os.path.join(path, 'alembic.ini'))

    captured_text = None
    def print_stdout(text, *arg):
        nonlocal captured_text
        captured_text = text

    alembic_cfg.print_stdout = print_stdout
    command.current(alembic_cfg)
    return captured_text