Skip to content

Instantly share code, notes, and snippets.

@siteshen
Created October 11, 2017 07:14
Show Gist options
  • Save siteshen/c9b1fd22d188e2ef53c4eee5ffe8619e to your computer and use it in GitHub Desktop.
Save siteshen/c9b1fd22d188e2ef53c4eee5ffe8619e to your computer and use it in GitHub Desktop.
SQLAlchemy keyset && offset/limit pagination query mixin
class PaginationMixin(object):
def paginate(self, order_by=None, order_dir=None, limit=None, cursor=None, page=None):
assert not page or not cursor, 'should not combine cusor and page'
# convert arguments
if isinstance(order_by, str):
order_by = getattr(self._primary_entity.expr._identity_class, order_by)
# check arguments
if (order_by is not None) and (not isinstance(order_by, InstrumentedAttribute)):
raise ValueError('Unrecognized type for order_by: %s' % type(order_by))
if order_dir not in {'asc', 'desc', None}:
raise ValueError('Unrecognized value for order_dir: %s' % order_dir)
if (limit is not None) and (limit <= 0):
raise ValueError('Unrecognized value for limit: %s' % order_dir)
query = self
if cursor is not None:
if order_dir == 'desc':
query = query.filter(order_by < cursor)
else:
query = query.filter(order_by > cursor)
if order_by is not None:
if order_dir == 'desc':
query = query.order_by(order_by.desc())
else:
query = query.order_by(order_by)
if page:
query = query.offset(limit * (page - 1))
if limit is not None:
query = query.limit(limit)
return query
def paginate_all(self, order_by='id', order_dir='asc', limit=10, cursor=None):
while True:
objects = self.paginate(order_by=order_by, order_dir=order_dir, limit=limit, cursor=cursor).all()
if objects:
yield objects, {'cursor': cursor}
cursor = getattr(objects[-1], order_by)
else:
break
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment