3

我正在开发一个使用 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但我想自动检测类型更改。有什么方法可以告诉 Alembicgeometry列的类型是Geometry并且完全没有变化？

4

1 回答 1

0

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

可以实现自定义compare_type函数并在env.py. 就我而言，geometry列被解释为sqlalchemy.Integer或sqalchemy.NUMERIC类型。这就是为什么我添加了一个 if 子句returns Falseif inspected_type is NUMERICorInteger和 metadata_type is 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问题无关。

于 2020-07-12T19:48:44.817 回答