quay/data/model/modelutil.py

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