How to check if there are pending migrations when using SQLAlchemy/Alembic?

Question

We're using SQLAlchemy and Alembic (along with Flask-SQLAlchemy and Flask-Migrate). How to check if there are pending migrations?

I tried to check both Alembic's and Flask-Migrate's documentation but failed to find the answer.

Answer 1

You can figure out if your project as at the latest migration with the current subcommand:

Example output when you are at the latest migration:

(venv) $ python app.py db current f4b4aa1dedfd (head)

The key thing is the (head) that appears after the revision number. That tells you that this is the most recent migration.

Here is how things change after I add a new migration, but before I upgrade the database:

(venv) $ python app.py db current f4b4aa1dedfd

And after I run db upgrade I get:

(venv) $ python app.py db current f3cd9734f9a3 (head)

Hope this helps!

Answer 2

What about using the alembic check command? Link

Answer 3

if you are running sqlaclhemy, alembic, with flask you can run

flask db check

and an example output would be No new upgrade operations detected.

Answer 4

Using @maciek's answer here is an async version:

import asyncio
from alembic import script
from alembic.runtime import migration
from src.database import get_db_connection_data
from sqlalchemy.ext.asyncio import create_async_engine


def check_pending_migrations(conn):
    script_ = script.ScriptDirectory('src/migrations')
    context = migration.MigrationContext.configure(conn)
    if context.get_current_revision() != script_.get_current_head():
        raise Exception('The database is not up-to-date. Please run your migrations.')

async def check_db():
    db_url = await get_db_connection_data()
    engine = create_async_engine(db_url)

    async with engine.connect() as conn:
        await conn.run_sync(check_pending_migrations)

    await engine.dispose()


if __name__ == '__main__':
    asyncio.run(check_db())

Answer 5

There is a tool for this https://github.com/4Catalyzer/alembic-autogen-check

Works like this

$ > alembic-autogen-check
INFO  [alembic.runtime.migration] Context impl PostgresqlImpl.
INFO  [alembic.runtime.migration] Will assume transactional DDL.
INFO  [alembic.autogenerate.compare] Detected NULL on column 'user.is_superuser'
ERROR: Migrations are out of sync with models. Diff:
[ [ ( 'modify_nullable',
      None,
      'user',
      'is_superuser',
      { 'existing_comment': None,
        'existing_server_default': False,
        'existing_type': BOOLEAN()},
      False,
      True)]]

You may need to run `PYTHONPATH=. alembic revision --autogenerate -m 'Your message'`.
 

Answer 6

Not 100% answering the question of the author but may add an additional option of how to do something in case of a pending alembic update.

We make a backup each time before alembic applies changes. We tried the previous proposals but preferred to add the backup call to the alembic template script.py.mako:

def upgrade():
    # create backup for recovery in case of failure
    backup.run()

    ${upgrades if upgrades else "pass"}

So everytime you create a migration script, e.g. with alembic revision ... the backup() call is added. The creator of the script may then even decide that for that current change there is no backup required, thus may remove those lines.

Answer 7

Here's how you can do it programatically:

from alembic import config
from alembic import script
from alembic.runtime import migration
import sqlalchemy

import exceptions


engine = sqlalchemy.create_engine(DATABASE_URL)
alembic_cfg = config.Config('alembic.ini')
script_ = script.ScriptDirectory.from_config(alembic_cfg)
with engine.begin() as conn:
    context = migration.MigrationContext.configure(conn)
    if context.get_current_revision() != script_.get_current_head():
        raise exceptions.DatabaseIsNotUpToDate('Upgrade the database.')

I've also published gist with this check.