物化视图的alembic生成答案

【问题标题】：alembic generation of materialized view物化视图的alembic生成
【发布时间】：2016-08-19 17:10:21
【问题描述】：

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

我正在使用flask-sqlalchemy，并且还使用带有postgres 的alembic。为了获得使用 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 都会知道如何更新视图？

非常感谢！

【问题讨论】：

出于好奇，你有没有想过这个？从今天开始走这条路……

标签： python flask sqlalchemy flask-sqlalchemy alembic

【解决方案1】：

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

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

两个重点：

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

【讨论】：