python - Alembic/Migrate 无法识别物化视图

Question

我postgres从这里按照 Jeff Windman 的方法创建了一个物化视图：

如何在插入对象后发出 SQLAlchemy 自定义 DDL？

和这里：

http://www.jeffwidman.com/blog/847/

视图工厂和模型已适应当前项目，但所有函数和类都是源模型的虚拟副本。该模型已导入视图并在模板中引用。遗憾的是，它被 Migrate(Alembic) 完全忽略，不迁移/升级。

用户.py

class User(UserMixin, db.Model):
    __tablename__ = 'users'
    id = db.Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
    username = db.Column(db.String(100), unique=True, nullable=False)
    password_hash = db.Column(db.String(255), nullable=False)
    individual_id = db.Column(UUID(as_uuid=True), db.ForeignKey('individuals.id', ondelete="CASCADE",
         onupdate="CASCADE"), nullable=False)
    role_id = db.Column(db.SmallInteger, db.ForeignKey('user_roles.id', ondelete='RESTRICT', 
         onupdate='CASCADE'), unique=False, index=True, nullable=False)
    mv_user_individual = db.relationship('User_individual_MV', backref='users', uselist=False, 
         primaryjoin='User.id==User_individual_MV.id', foreign_keys='User_individual_MV.id')

个人.py

class Individual(db.Model):
    __tablename__ = 'individuals'
    id = db.Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
    first_name = db.Column(db.String(100), unique=False, index=False, nullable=False)
    last_name = db.Column(db.String(100), unique=False, index=False, nullable=False)
    users = db.relationship('User', 
        lazy='subquery', 
        backref=db.backref('individuals', lazy='select'))

mat_view_factory.py

from sqlalchemy.ext import compiler
from sqlalchemy.schema import DDLElement, PrimaryKeyConstraint
from app import db


class CreateMaterializedView(DDLElement):
    def __init__(self, name, selectable):
        self.name = name
        self.selectable = selectable


@compiler.compiles(CreateMaterializedView)
def compile(element, compiler, **kw):
    # Could use "CREATE OR REPLACE MATERIALIZED VIEW..."
    # but I'd rather have noisy errors
    return 'CREATE MATERIALIZED VIEW %s AS %s' % (
        element.name,
        compiler.sql_compiler.process(element.selectable, literal_binds=True),
        )


def create_mat_view(name, selectable, metadata=db.metadata):
    _mt = db.MetaData()  # temp metadata just for initial Table object creation
    t = db.Table(name, _mt)  # the actual mat view class is bound to db.metadata
    for c in selectable.c:
        t.append_column(db.Column(c.name, c.type, primary_key=c.primary_key))

    if not (any([c.primary_key for c in selectable.c])):
        t.append_constraint(PrimaryKeyConstraint(*[c.name for c in selectable.c]))

    db.event.listen(
        metadata, 'after_create',
        CreateMaterializedView(name, selectable)
        )

    @db.event.listens_for(metadata, 'after_create')
    def create_indexes(target, connection, **kw):
        for idx in t.indexes:
            idx.create(connection)

    db.event.listen(
        metadata, 'before_drop',
        db.DDL('DROP MATERIALIZED VIEW IF EXISTS ' + name)
        )
    return t


def refresh_mat_view(name, concurrently):
    # since session.execute() bypasses autoflush, must manually flush in order
    # to include newly-created/modified objects in the refresh
    db.session.flush()
    _con = 'CONCURRENTLY ' if concurrently else ''
    db.session.execute('REFRESH MATERIALIZED VIEW ' + _con + name)


def refresh_all_mat_views(concurrently=True):
    '''Refreshes all materialized views. Currently, views are refreshed in
    non-deterministic order, so view definitions can't depend on each other.'''
    mat_views = db.inspect(db.engine).get_view_names(include='materialized')
    for v in mat_views:
        refresh_mat_view(v, concurrently)


class MaterializedView(db.Model):
    __abstract__ = True

    @classmethod
    def refresh(cls, concurrently=True):
        '''Refreshes the current materialized view'''
        refresh_mat_view(cls.__table__.fullname, concurrently)

User_individual_MV.py：

from app import db
from app.mat_view_factory import MaterializedView, create_mat_view
from app.models import User, Individual

class User_individual_MV(MaterializedView):
    __table__ = create_mat_view('user_individual_mv',
                    db.select(
                        [User.id.label('id'),
                        User.username.label('username'),
                        User.role_id.label('role_id'),
                        Individual.id.label('individual_id'),
                        Individual.first_name.label('first_name'),
                        Individual.last_name.label('last_name')
                        ]
                    ).select_from(db.join(User, Individual, isouter=False))
                    ).group_by(Individual.last_name)
    )
db.Index('uq_user_individual_mv', User_individual_MV.id, unique=True)

任何建议将不胜感激。

标签： pythonpostgresqlflask-sqlalchemyalembicflask-migrate