This repository has been archived on 2020-03-24. You can view files and clone it, but cannot push or open issues or pull requests.
quay/data/model/modelutil.py
Joseph Schorr b0b7b63be9 Fix queue tests for MySQL
MySQL's date time's appear to have a 1 second threshold, so we need to make sure the queue items added for the tests are available as soon as they are added. Before this change, the available_after was set to `datetime.utcnow()`, and, if the `get` was called within 1 second, then its check would fail.
2016-07-15 13:27:50 -04:00

38 lines
1.3 KiB
Python

from peewee import SQL
def paginate(query, model, descending=False, page_token=None, limit=50, id_alias=None):
""" Paginates the given query using an ID range, starting at the optional page_token.
Returns a *list* of matching results along with an unencrypted page_token for the
next page, if any. If descending is set to True, orders by the ID descending rather
than ascending.
"""
# Note: We use the id_alias for the order_by, but not the where below. The alias is necessary
# for certain queries that use unions in MySQL, as it gets confused on which ID to order by.
# The where clause, on the other hand, cannot use the alias because Postgres does not allow
# aliases in where clauses.
id_field = model.id
if id_alias is not None:
id_field = SQL(id_alias)
if descending:
query = query.order_by(id_field.desc())
else:
query = query.order_by(id_field)
if page_token is not None:
start_id = page_token.get('start_id')
if start_id is not None:
if descending:
query = query.where(model.id <= start_id)
else:
query = query.where(model.id >= start_id)
results = list(query)
page_token = None
if len(results) > limit:
start_id = results[limit].id
page_token = {
'start_id': start_id
}
return results[0:limit], page_token