SQLAlchemy func.count with filter

Question

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

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

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

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?

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

r-m-n · Accepted Answer · 2015-12-24 12:54:04Z

6

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

edited Dec 24, 2015 at 12:54

answered Dec 24, 2015 at 8:53

r-m-n

15.3k6 gold badges74 silver badges71 bronze badges

Sign up to request clarification or add additional context in comments.

2 Comments

fasouto Over a year ago

Thanks for replying but this wont work for me. The query is constructed dynamically, is not all the time the same.

fasouto Over a year ago

Thank you very much! I was stucked with this simple problem for hours! :)

Collectives™ on Stack Overflow

SQLAlchemy func.count with filter

1 Answer 1

2 Comments

Your Answer

Linked

Hot Network Questions

Collectives™ on Stack Overflow

1 Answer 1

2 Comments

Your Answer

Sign up or log in

Post as a guest

Linked

Related