54 lines
1.8 KiB
Python
54 lines
1.8 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)
|
|
|
|
start_id = pagination_start(page_token)
|
|
if start_id is not None:
|
|
if descending:
|
|
query = query.where(model.id <= start_id)
|
|
else:
|
|
query = query.where(model.id >= start_id)
|
|
else:
|
|
query = query.limit(limit + 1)
|
|
|
|
return paginate_query(query, limit=limit, id_alias=id_alias)
|
|
|
|
|
|
def pagination_start(page_token=None):
|
|
""" Returns the start ID for pagination for the given page token. Will return None if None. """
|
|
if page_token is not None:
|
|
return page_token.get('start_id')
|
|
|
|
return None
|
|
|
|
|
|
def paginate_query(query, limit=50, id_alias=None):
|
|
""" Executes the given query and returns a page's worth of results, as well as the page token
|
|
for the next page (if any).
|
|
"""
|
|
results = list(query)
|
|
page_token = None
|
|
if len(results) > limit:
|
|
start_id = getattr(results[limit], id_alias or 'id')
|
|
page_token = {
|
|
'start_id': start_id
|
|
}
|
|
|
|
return results[0:limit], page_token
|