SQLAlchemy func.count с фильтром

Я использую структуру, которая выполняет разбиение на страницы следующим образом:

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
person fasouto    schedule 24.12.2015    source источник

Ответы (1)


arrow_upward
6
arrow_downward

вы можете использовать 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())
person r-m-n    schedule 24.12.2015
comment
Спасибо за ответ, но у меня это не сработает. Запрос строится динамически, не всегда одинаков. - person fasouto; 24.12.2015
comment
Большое спасибо! Я на несколько часов застрял над этой простой проблемой! :) - person fasouto; 24.12.2015