SQLAlchemy - 你可以向查询对象添加自定义方法吗？答案

【问题标题】：SQLAlchemy - can you add custom methods to the query object?SQLAlchemy - 你可以向查询对象添加自定义方法吗？
【发布时间】：2013-04-10 20:50:21
【问题描述】：

有没有办法为查询对象创建自定义方法，以便您可以执行类似的操作？

User.query.all_active()

all_active() 本质上是.filter(User.is_active == True)

并且能够过滤掉它？

User.query.all_active().filter(User.age == 30)

【问题讨论】：

我想知道其他question/answer 中的任何内容是否对此有用？（不过，它与其说是一种自定义方法……倒不如说是一种自定义动态过滤器。）

标签： python sqlalchemy

【解决方案1】：

您可以继承 Query 基类来添加您自己的方法：

from sqlalchemy.orm import Query

class MyQuery(Query):

  def all_active(self):
    return self.filter(User.is_active == True)

然后在创建会话时告诉 SQLAlchemy 使用这个新的查询类 (docs here)。从您的代码看来，您可能正在使用 Flask-SQLAlchemy，因此您可以按如下方式进行：

db = SQLAlchemy(session_options={'query_cls': MyQuery})

否则，您会将参数直接传递给sessionmaker：

sessionmaker(bind=engine, query_cls=MyQuery)

到目前为止，这个新的查询对象并不那么有趣，因为我们在方法中硬编码了User 类，所以它不能用于其他任何东西。更好的实现将使用查询的底层类来确定应用哪个过滤器。这有点棘手，但也可以做到：

class MyOtherQuery(Query):

  def _get_models(self):
    """Returns the query's underlying model classes."""
    if hasattr(query, 'attr'):
      # we are dealing with a subquery
      return [query.attr.target_mapper]
    else:
      return [
        d['expr'].class_
        for d in query.column_descriptions
        if isinstance(d['expr'], Mapper)
      ]

  def all_active(self):
    model_class = self._get_models()[0]
    return self.filter(model_class.is_active == True)

最后，动态关系（如果有的话）不会使用这个新的查询类。为了让那些也使用它，您可以在创建关系时将其作为参数传递：

users = relationship(..., query_class=MyOtherQuery)

【讨论】：

这对我不起作用。 gist.github.com/nickretallack/76bcd65cc0305abcd33b
@NickRetallack 我在创建 BaseModel 类和声明 query_class 类属性方面取得了一些成功。但是，这似乎是问题所在：github.com/mitsuhiko/flask-sqlalchemy/blob/…
所以你说一个会话只能与一个查询类一起工作。我们能做得更好吗？并允许同一个会话动态选择查询类？我的意思是不同的表可以有不同的查询类，但我可能想在同一个会话中使用它们（例如在一个事务中使用它们时）。
我觉得有错别字，不应该是query._column_descriptions（带下划线）吗？也可能添加导入，会有所帮助，即使它似乎全部来自 orm 包。

【解决方案2】：

这对我来说很好用

class ParentQuery(Query):
    def _get_models(self):     
        if hasattr(query, 'attr'):
            return [query.attr.target_mapper]
        else:
            return self._mapper_zero().class_

    def FilterByCustomer(self):
        model_class = self._get_models()
        return self.filter(model_class.customerId == int(g.customer.get('customerId')))


class AccountWorkflowModel(db.Model):
    query_class = ParentQuery
    .................

【讨论】：

能否请您显示完整的代码。我想在 sqlalchemy 中添加自定义过滤器，但我也希望它在相关模型中初始化（仅在模型类中进行编码，并且不想像会话和引擎等那样接触 sqlalchemy conf）。我想写这种代码stackoverflow.com/a/56806069/5470263 好像可以用你的方法实现。
嗨，这是完整的代码。你哪里有问题？请告诉我，我可以发送您需要的指定代码。用法：@staticmethod def getAllItems()：返回 AccountWorkflowModel.query.FilterByCustomer().all()

【解决方案3】：

要提供一个自定义方法，该方法将被所有继承自特定父级的模型使用，首先如前所述从 Query 类继承：

from flask_sqlalchemy import SQLAlchemy, BaseQuery
from sqlalchemy.inspection import inspect

class MyCustomQuery(BaseQuery):
    def all_active(self):
        # get the class
        modelClass = self._mapper_zero().class_
        # get the primary key column
        ins = inspect(modelClass)
        # get a list of passing objects
        passingObjs = []
        for modelObj in self:
            if modelObj.is_active == True:
                # add to passing object list
                passingObjs.append(modelObj.__dict__[ins.primary_key[0].name])
        # change to tuple
        passingObjs = tuple(passingObjs)
        # run a filter on the query object
        return self.filter(ins.primary_key[0].in_(passingObjs))

# add this to the constructor for your DB object
myDB = SQLAlchemy(query_class=MyCustomQuery)

这是针对flask-sqlalchemy的，人们在寻找这个答案时仍然会来到这里。

【讨论】：