SQLAlchemy func.count z filtrem

Question

Używam ramy, które wykonuje paginacji tak:

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

chcę zastąpić metodę get_count_query użyć własnego zapytanie bo jestem filtrowania pewne rezultaty i get_count_query tylko zwraca wszystkie elementy w tabeli. Zapytania są tworzone dynamicznie, na przykład jedno zapytanie może być:

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

mogę policzyć elementy tego zapytania łatwo 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() 

Ale to zawiedzie raz to osiągnąć .scalar() metoda (i nie mogę usunąć tej metody). Pytanie brzmi: jak mogę zastosować func.count('*') do istniejącego zapytania?

Czy mogę pobrać filtry z mojego zapytania i zastosować je do zapytania func.count('*')?

Answer 1

można użyć select_from z join i 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))) 

z podzapytanie

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