物化视图的蒸馏生成

Question

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

我正在使用flask-sqlalchemy，还使用alembic和postgres。为了获得与 sqlalchemy 一起使用的物化视图，我关注了关于该主题的nice post。我大量使用它，只有一些细微的差异（本文也使用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

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

编辑：现在可能有一种方法可以自动生成视图迁移。查看答案https://stackoverflow.com/a/72829474/2839862

我认为解决这个问题最简单的方法是不要依赖 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

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

activities = table(
    "activities",
    sa.Column("id", sa.Integer()),
    ...
)

view_query = (
    select(
        [
            activities.c.id,
        ]
    )
    .select_from(activities)
)

def upgrade():
    view_query_string = str(view_query.compile(compile_kwargs={"literal_binds": True}))

    op.execute(f"CREATE VIEW activity_view AS {view_query_string}")

def downgrade():
    op.execute("DROP VIEW activity_view")

重要的两点：

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

Answer 2

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

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

from alembic_utils.pg_materialized_view import PGMaterializedView

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

您可以将

definition

基于静态 SQL 或 sqlalchemy 代码的编译版本。

然后在你的alembic

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了解更多信息。