带有过滤器的 SQLAlchemy func.count答案

【问题标题】：SQLAlchemy func.count with filter带有过滤器的 SQLAlchemy func.count
【发布时间】：2017-06-13 00:24:55
【问题描述】：

我正在使用这样的分页框架：

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('*') 查询吗？

【问题讨论】：

标签： python sqlalchemy flask-sqlalchemy

【解决方案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())

【讨论】：

感谢您的回复，但这对我不起作用。查询是动态构造的，并不总是一样的。
非常感谢！我被这个简单的问题困扰了好几个小时！ :)