SQLAlchemy func.count with filter

Question

I use a framework that performs pagination as follows:

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() ... 

I want to override the get_count_query method to use my own query, because I filter some results, and get_count_query returns all the elements in the table. Requests are created dynamically, for example, one request can be:

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

I can easily count the elements in this query using 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() 

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

Is it possible to get filters from my query and apply them to func.count('*') ?

Answer 1

you can use select_from with join and 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))) 

with subquery

 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())