postgresql - 使用 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)存在。