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