使用 SQLAlchemy/Alembic 时如何检查是否有挂起的迁移？

Question

我们正在使用 SQLAlchemy 和 Alembic（以及 Flask-SQLAlchemy 和 Flask-Migrate）。如何检查是否有待处理的迁移？

我尝试检查 Alembic 和 Flask-Migrate 的文档，但未能找到答案。

Answer 1

您可以使用

current

子命令确定您的项目是否处于最新迁移状态：

最近一次迁移时的示例输出：

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

关键是修订号后面出现的

(head)

。这告诉您这是最近的迁移。

以下是添加新迁移后、升级数据库之前情况发生的变化：

(venv) $ python app.py db current
f4b4aa1dedfd

我跑步后

db upgrade

我得到：

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

希望这有帮助！

Answer 2

以下是如何以编程方式完成此操作：

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

我还发布了这张支票的要点。

Answer 3

有一个工具用于此https://github.com/4Catalyzer/alembic-autogen-check

像这样工作

$ > 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 4

这里使用@maciek的答案是异步版本：

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

使用

alembic check

命令怎么样？ 链接

Answer 6

不是 100% 回答作者的问题，但可能会添加一个额外的选项，说明如何在等待 alembic 更新的情况下执行某些操作。

每次在 alembic 应用更改之前我们都会进行备份。我们尝试了之前的建议，但更愿意将备份调用添加到 alembic 模板

script.py.mako

:

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

    ${upgrades if upgrades else "pass"}

因此每次创建迁移脚本时，例如使用

alembic revision ...

添加 backup() 调用。然后，脚本的创建者甚至可以决定对于当前的更改不需要备份，因此可以删除这些行。

Answer 7

如果你正在运行 sqlaclhemy、alembic，则可以使用 Flask 运行

flask db check

示例输出为

No new upgrade operations detected.