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.

Flintshire

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!

Ensor

This is not what I see: # flask db current gives me INFO [alembic.runtime.migration] Context impl MySQLImpl. INFO [alembic.runtime.migration] Will assume non-transactional DDL. – Regent

Seems like when migrations are up to date 8ca18d2d21bf (head) is appended to the above – Regent

This should be the accept answer – Randeerandel

Answer 2

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.

Cheremkhovo

This comment is/was super helpful. For others that come here, it may help to use the path to the project's alembic folder, rather than rely on the .ini file. script.ScriptDirectory('/path/to/alembic/project') – Heine

Answer 3

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

Lianneliao

Answer 4

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'`.
 

Orgel

Answer 5

What about using the alembic check command? Link

Steeplechase

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.

Anette

Answer 7

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.

Rennarennane