使用 SQLAlchemy 和 Alembic 检查数据库中是否存在表列

Question

我正在使用 Alembic 作为迁移工具，并且我正在一个已经更新的数据库上启动以下伪脚本（Alembic 没有修订条目，数据库架构是最新的）。

revision = '1067fd2d11c8'
down_revision = None

from alembic import op
import sqlalchemy as sa


def upgrade():
    op.add_column('box', sa.Column('has_data', sa.Boolean, server_default='0'))


def downgrade():
    pass

它只在 PostgreSQL 后面给我以下错误（这对 MySQL 来说都很好）：

INFO  [alembic.migration] Context impl PostgresqlImpl.
INFO  [alembic.migration] Will assume transactional DDL.
INFO  [root] (ProgrammingError) ERREUR:  la colonne « has_data » de la relation « box » existe déjà

最后一行表示该列has_data已存在。

我想检查该列之前是否存在op.add_column。

Answer 1

我们遇到了同样的问题：当架构中可能存在修订版中添加的列时，我们必须适应边缘情况。不能选择消除错误，因为这将回滚当前事务（除非使用 sqlite），并且版本表不会更新。检查列是否存在在这里似乎是最佳选择。这是我们的解决方案（与已接受的答案中的想法相同，但针对 2022 年进行了更新）：

from alembic import op
from sqlalchemy import inspect

def column_exists(table_name, column_name):
    bind = op.get_context().bind
    insp = inspect(bind)
    columns = insp.get_columns(table_name)
    return any(c["name"] == column_name for c in columns)

op.get_context()这是从修订文件中调用的，因此已配置了通过访问的上下文（大概在您的env.py, 并且bind存在。

Answer 2

不幸的是，我面临的情况是，我们有多个具有不同模式的版本，所有这些版本都需要迁移到单个代码库。目前还没有任何迁移，并且任何数据库中都没有标记版本。因此，第一次迁移将进行这些条件检查。第一次迁移后，一切都将处于已知状态，我可以避免此类黑客攻击。

所以我在迁移中添加了这个（信用属于http://www.derstappen-it.de/tech-blog/sqlalchemie-alembic-check-if-table-has-column）：

from alembic import op
from sqlalchemy import engine_from_config
from sqlalchemy.engine import reflection

def _table_has_column(table, column):
    config = op.get_context().config
    engine = engine_from_config(
        config.get_section(config.config_ini_section), prefix='sqlalchemy.')
    insp = reflection.Inspector.from_engine(engine)
    has_column = False
    for col in insp.get_columns(table):
        if column not in col['name']:
            continue
        has_column = True
    return has_column

我的升级功能具有以下检查（请注意，我设置了一个批处理标志来添加该with op.batch_alter_table行，这可能不在大多数设置中：

def upgrade():
    # ### commands auto generated by Alembic - please adjust! ###
    with op.batch_alter_table('mytable', schema=None) as batch_op:
        if not _table_has_column('mytable', 'mycol'):
            batch_op.add_column(sa.Column('mycol', sa.Integer(), nullable=True))
        if not _table_has_column('mytable', 'mycol2'):
            batch_op.add_column(sa.Column('mycol2', sa.Integer(), nullable=True))

Answer 3

最简单的答案是不要尝试这样做。相反，让您的 Alembic 迁移代表数据库的完整布局。然后您所做的任何迁移都将基于对现有数据库的更改。

如果您已经有一个数据库，要开始迁移，请暂时指向一个空数据库并运行alembic revision --autogenerate -m "base". 然后，指向实际的数据库并运行alembic stamp head以说的当前状态由最新的迁移表示，而不实际运行它。

如果您出于某种原因不想这样做，您可以选择不使用--autogenerate，而是生成用您想要的操作填充的空修订。Alembic 不会阻止您这样做，只是不那么方便。