python - SQLAlchemy func.count 带过滤器

Question

我正在使用一个像这样进行分页的框架:

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 只返回表中的所有元素。查询是动态创建的，例如一个查询可以是:

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

我可以使用 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('*') 应用于现有查询？

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

Answer 1

您可以将 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())