Flask-SQLAlchemy：'deleted' 个条目的自动过滤器

Question

我正在开发一个庞大的代码库，它执行以下所有数据库交互：

ExampleClass.query.all()

其中 ExampleClass 继承了 db.Model 的属性，我们的 BaseEntity 定义如下：

class BaseEntity(object):
    deleted = db.Column(db.DateTime, default=None, nullable=True)
    ... # more columns

现在，当我查询 ExampleClass 的所有条目时，ExampleClass.query.all()，我希望删除字段设置为日期的条目，也就是从系统中删除的条目被排除在外.如果我不必用一个简单的 .filter(deleted != None) 更新整个代码库，那将是理想的。我的解决方案是添加一个 SQLAlchemy 事件过滤器：before_compile。文档准确地告诉了我要搜索的内容：

@event.listens_for(Query, "before_compile", retval=True)
def no_deleted(query):
    for desc in query.column_descriptions:
        if desc['type'] is ExampleClass:
            entity = desc['expr']
            query = query.filter(entity.deleted == False)
    return query

但是我无法让它工作，我遇到的最后一个错误如下：

AttributeError: 'Mapper' object has no attribute 'deleted'

我使用的库版本是：Flask==0.10.1, Flask-SQLAlchemy==1.0, SQLAlchemy==1.0.6

Answer 1

执行此操作的明显且简单的方法（经过测试并适用于我）是：

class BaseEntity(db.Model):
    deleted = db.Column(db.DateTime, default=None, nullable=True)

BaseEntity.query = BaseEntity.query.filter(BaseEntity.deleted == None)

也就是说，其他开发人员可能希望 BaseEntity.query 以干净的查询对象开头，因此您可以考虑使用与 query 不同的 class 属性:

class BaseEntity(db.Model):
    deleted = db.Column(db.DateTime, default=None, nullable=True)

BaseEntity.non_deleted = BaseEntity.query.filter(BaseQuery.deleted == None)

# and then use it like …
BaseEntity.non_deleted.all()

Answer 2

我知道这是个老问题，但最近我用这个函数让它工作：

@event.listens_for(Query, "before_compile", retval=True) 
def no_deleted(query):
    for desc in query.column_descriptions:
        if desc['type'] is ExampleClass:
           entity = desc['entity']

           limit, offset = query._limit, query._offset
           query._limit, query._offset = None, None

           query = query.filter(entity.deleted == False)

           query._limit, query._offset = limit, offset

    return query

您需要对实体而非类型进行过滤。并临时删除限制和偏移量。

Flask==1.0.2，Flask-SQLAlchemy==2.3.2，SQLAlchemy==1.2.12

link to sqlalchemy docs for this event

升级版本：

如果您不想排除已删除的项目，可以使用此选项：

创建 QueryMixin：

class ShowDeleted:
    def show_deleted(self):
        q = self._clone()
        q.__dict__['__show_deleted'] = True
        return q

编辑no_deleted函数

def no_deleted(query):
    for desc in query.column_descriptions:
        show_deleted = query.__dict__.get('__show_deleted', False)
        if not show_deleted and desc['type'] is ExampleClass:
            entity = desc['entity']

            limit, offset = query._limit, query._offset
            query._limit, query._offset = None, None

            query = query.filter(entity.deleted == False)

            query._limit, query._offset = limit, offset

    return query

如果您使用的是 FlaskSQLAlchemy，请按以下方式设置数据库：

from flask_sqlalchemy import SQLAlchemy, BaseQuery

db = SQLAlchemy(query_class= type('Query', (ShowDeleted, BaseQuery), {}))

如果是原始 sqlalchemy，请使用此代码段：

from sqlalchemy.orm import Query
db = scoped_session(sessionmaker(query_cls=type('Query', (ShowDeleted, Query), {})))

现在您可以在您的代码中调用

db.session.query(ExampleClass).show_deleted().all()

Flask-SQLAlchemy：'deleted' 个条目的自动过滤器

Flask-SQLAlchemy: Automatic filter for 'deleted' entries

sqlalchemy

flask

python-3.x

flask-sqlalchemy

升级版本：