带过滤器的SQLAlchemy func.count [英] SQLAlchemy func.count with filter

问题描述

我使用的是这样的分页框架:

I'm using a framework that does pagination like this:

def get_count_query(self):
    return self.session.query(func.count('*')).select_from(self.model)

def paginate(self):
    ... <irrelevant>...
    count = self.get_count_query.scalar()
    ...

我想重写get_count_query方法以使用我自己的查询，因为我正在过滤一些结果，而get_count_query只是返回表中的所有元素.查询是动态创建的，例如，一个查询可以是:

I want to override the get_count_query method to use my own query because I'm filtering some results and get_count_query just returns all elements in the table. Queries are created dynamically, for example one query could be:

Asset.query.join(StatusLabel).filter(StatusLabel.status == 'Deployable', or_(
                                     Asset.assigned_to.isnot(None)),
                                     Asset.deleted_at.is_(None))

我可以使用query.count()轻松计算该查询中的元素:

I can count the elements in this query easily with query.count():

def get_count_query(self):
    q = Asset.query.join(StatusLabel).filter(StatusLabel.status == 'Deployable', or_(
                                             Asset.assigned_to.isnot(None)),
                                             Asset.deleted_at.is_(None))
    return q.count()

但是一旦到达.scalar()方法，该操作将失败(并且我无法删除此方法).所以问题是:如何将func.count('*')应用于现有查询?

But this will fail once it reach the .scalar() method (and I cannot remove this method). So the question is: how can I apply func.count('*') to an existing query?

我可以从查询中检索过滤器并将其应用于func.count('*')查询吗?

Can I retrieve the filters from my query and apply them to the func.count('*') query?

推荐答案

您可以将select_from与join和filter

def get_count_query(self):
    return self.session.query(func.count('*')).select_from(Asset).join(StatusLabel)\
                                              .filter(StatusLabel.status == 'Deployable', or_(
                                                      Asset.assigned_to.isnot(None),
                                                      Asset.deleted_at.is_(None)))

带有子查询

def get_count_query(self):
    q = Asset.query.join(StatusLabel).filter(StatusLabel.status == 'Deployable', or_(
                                             Asset.assigned_to.isnot(None)),
                                             Asset.deleted_at.is_(None))

    return self.session.query(func.count('*')).select_from(q.subquery())

这篇关于带过滤器的SQLAlchemy func.count的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！