alembic生成物化视图

Question

TL; DR:我如何让sqmbic理解并为sqlalchemy中创建的物化视图生成SQL？

我正在使用flask-sqlalchemy并使用alembic和postgres.为了获得使用sqlalchemy的物化视图,我在该主题上发表了一篇很好的帖子.我使用它很多,只有一些小的分歧(文章也使用flask-sqlalchemy,但完整的代码示例直接使用sqlalchemy的声明性基础).

class ActivityView(db.Model):
    __table__ = create_materialized_view(
        'activity_view',
        db.select([
            Activity.id.label('id'),
            Activity.name.label('name'),
            Activity.start_date.label('start_date'),
        ]).where(
            db.and_(
                Activity.start_date != None,
                Activity.start_date <=
                    datetime_to_str(datetime.now(tz=pytz.UTC) + timedelta(hours=48))
            )
        )
    )

    @classmethod
    def refresh(cls, concurrently=True):
        refresh_materialized_view(cls.__table__.fullname, concurrently)

db.Index('activity_view_index',
         ActivityView.__table__.c.id, ActivityView.__table__.c.start_date,
         unique=True)

这些create_materialized_view和refresh_materialized_view方法直接来自博客文章.

请注意,上面的示例已经大大简化,并且由于我的简化可能看起来很愚蠢,但我想要了解的真正想法是如何在迁移期间将alembic转换为一系列的alembic操作？

当我运行测试时,代码运行正常,视图生成正常,一切正常.当alembic运行时,它不会对视图执行任何操作.所以我最终要做的是将测试为物化视图发出的SQL复制到alembic迁移/版本文件中,然后最终直接执行:

op.execute(activities_view_sql)

类似地,我在物化视图上生成唯一索引时执行相同的直接SQL执行.

不幸的是,我的方法容易出错,并且会产生看似不必要的代码重复.

有没有办法让alembic了解我,ActivityView以便任何时候它改变,alembic将知道如何更新视图？

非常感谢!

Answer 1

虽然这个问题没有具体指出使用 PostgreSQL，但帖子说它是基于 PostgeSQL 中的目标物化视图，所以这个答案还针对一个名为alembic_utils的附加包，它基于alembic ReplaceableObjects，增加了对自动生成更大的支持PostgreSQL 实体类型的数量，包括函数、视图、物化视图、触发器和策略。

要进行设置，您可以通过以下方式创建物化视图；

from alembic_utils.pg_materialized_view import PGMaterializedView

actview = PGMaterializedView (
    schema="public",
    signature="activity_view",
    definition="select ...",
    with_data=True
)

您可以基于definition静态 SQL 或 sqlalchemy 代码的编译版本。

然后在你的蒸馏器 env.py中：

from foo import actview
from alembic_utils.replaceable_entity import register_entities

register_entities([actview])

当物化视图在代码中更新时，Alembic 现在将自动生成迁移。

编辑：10/10/2023 - 应该注意的是，alemic_utils 不会自动管理物化视图上的索引，这些需要手动管理。请参阅： https: //github.com/olirice/alembic_utils/pull/46了解更多信息。

Answer 2

TLDR：只需手动编写视图迁移即可。似乎没有对视图自动生成的合理支持。

编辑：现在可能有一种方法可以自动生成视图迁移。查看答案/sf/answers/5098063211/

我认为解决这个问题最简单的方法是不要依赖 Alembic 为您自动生成视图。相反，您可以指示它在 Alembic 中忽略这样的视图env.py：

def include_object(obj, name, type_, reflected, compare_to):
    if obj.info.get("is_view", False):
        return False

    return True

...

def run_migrations_offline():
...
    context.configure(url=url, target_metadata=target_metadata, literal_binds=True, include_object=include_object)

...

def run_migrations_online():
    ....
    with connectable.connect() as connection:
        context.configure(connection=connection, target_metadata=target_metadata, include_object=include_object)

该标志由我的自定义基类is_view设置：View

class View(Model):
    @classmethod
    def _init_table(cls, sub_cls):
        table: sa.Table = Model._init_table(sub_cls)

        if table is None:
            return table

        table.info["is_view"] = True
        return table

当自动生成忽略视图时，您可以手动将适当的命令添加到迁移中：

class View(Model):
    @classmethod
    def _init_table(cls, sub_cls):
        table: sa.Table = Model._init_table(sub_cls)

        if table is None:
            return table

        table.info["is_view"] = True
        return table

两个要点：

1. 代码重复并不总是一件坏事 - 您可以将迁移视为版本控制工具，而不是常规代码。您的版本历史记录不应取决于代码库的当前状态
2. 手动编写的迁移可以说比生成的迁移更容易出错，但是您可以通过在生产应用程序的测试中运行迁移来部分缓解这一问题。此外，仅检查生成的数据库模式应该会有所帮助。