Alembic 为 Geometry 列生成任意类型更改

Question

我正在开发一个使用 SQLite 作为数据库并使用 Alembic 作为数据库迁移工具的项目。它包含空间数据，因此项目中包含空间扩展和geoalchemy2。我正在使用autogenerate 命令，它检测到几何列中不存在的一些更改。

这是项目的简化结构：

    # Model
    sqlite_naming_convention = {
    "ix": "ix_%(column_0_label)s",
    "uq": "uq_%(table_name)s_%(column_0_name)s",
    "ck": "ck_%(table_name)s_%(column_0_name)s",
    "fk": "fk_%(table_name)s_%(column_0_name)s_%(referred_table_name)s",
    "pk": "pk_%(table_name)s",
    }
    Metadata = MetaData(naming_convention=sqlite_naming_convention)
    BaseSpatiaLite = declarative_base(metadata=Metadata)


    class Geometries(BaseSpatiaLite):
        __tablename__ = "Geometries"

        geometry_id = Column(Integer, primary_key=True)
        geometry = Column(
            geoalchemy2.types.Geometry(geometry_type="GEOMETRY", srid=4326, management=True),
            nullable=False,
        )
        name = Column(String(length=150), nullable=False)

Alembic的env.py如下：

    # env.py
    ...
    def run_migrations_online():
        connectable = engine_from_config(
            config.get_section(config.config_ini_section),
            prefix="sqlalchemy.",
            poolclass=pool.NullPool,
        )
        # Enables Spatialite extension
        listen(connectable, "connect", load_spatialite)
        # Creates Spatial tables if they don't exist
        create_spatial_tables_for_sqlite(connectable)
        with connectable.connect() as connection:
            context.configure(
                connection=connection,
                target_metadata=target_metadata,
                render_as_batch=True,
                compare_type=True,
            )
    
            with context.begin_transaction():
                context.run_migrations()

创建几何表的第一个迁移脚本：

    ...
    def upgrade():
        op.create_table(
            "Geometries",
            sa.Column("geometry_id", sa.Integer(), nullable=False),
            sa.Column("geometry", geoalchemy2.types.Geometry(management=True), nullable=False),
            sa.Column("name", sa.String(length=150), nullable=False),
            sa.PrimaryKeyConstraint("geometry_id"),
        )
    
    
    def downgrade():
        op.drop_table(
            "Geometries",
        )

运行此迁移脚本后，表已正确创建：

当我再次运行autogenerate 命令时，它应该没有发现任何变化。但是，它会生成一个具有任意类型更改的迁移脚本：

    def upgrade():
        with op.batch_alter_table("Geometries", schema=None) as batch_op:
            batch_op.alter_column(
                "geometry",
                existing_type=sa.NUMERIC(),
                type_=geoalchemy2.types.Geometry(srid=4326, management=True),
                nullable=False,
            )
    
    
    def downgrade():
        with op.batch_alter_table("Geometries", schema=None) as batch_op:
            batch_op.alter_column(
                "geometry",
                existing_type=geoalchemy2.types.Geometry(srid=4326, management=True),
                type_=sa.NUMERIC(),
                nullable=True,
            )

我知道我可以将compare_type 参数设置为False，但我想自动检测类型更改。有没有办法告诉Alembic geometry 列的类型是Geometry 并且完全没有变化？

Answer 1

我找到了解决方案。我在这里分享它以防其他人可能会遇到此错误：(https://alembic.sqlalchemy.org/en/latest/autogenerate.html#comparing-types)

可以实现自定义compare_type 函数并在env.py 中使用它。在我的例子中，geometry 列被解释为sqlalchemy.Integer 或sqalchemy.NUMERIC 类型。这就是为什么我添加了一个 if 子句，其中 returns False if inspected_type 是 NUMERIC 或 Integer 并且 metadata_type 是 geoalchemy2.types.Geometry。

# add it to env.py
def custom_compare_type(
    context, 
    inspected_column, 
    metadata_column, 
    inspected_type, 
    metadata_type
):
# return False if the metadata_type is the same as the inspected_type
# or None to allow the default implementation to compare these
# types. a return value of True means the two types do not
# match and should result in a type change operation.
if (isinstance(inspected_type, NUMERIC) or isinstance(inspected_type, Integer)) and isinstance(
    metadata_type, Geometry
):
    return False

return None

当您将compare_type=True 更改为compare_type=custom_compare_type 时，Alembic 应该停止检测geometry 列的任意类型更改！

注意：Alembic 仍检测到可空性更改，但与compare_type 问题无关。