Add pagination helpers

[senko]

Django's built-in paginator works in terms of fixed-size pages. While usable in API context, this might pose problems if results are likely to change between subsequent page requests (eg. more data pours in).

A simple pagination model with since (all objects after the selected one) and limit works better in this case. Here's a sample implementation:

def get_paginated_data(data, start, count, order='-id'):
    if order.startswith('-'):
        asc = False
        order_field = order[1:]
    else:
        asc = True
        order_field = order

    ordered_data = data.order_by(order)
    if start:
        filter_fields = {
            order_field + ('__gte' if asc else '__lte'): start
        }
        ordered_data = ordered_data.filter(**filter_fields)

    retval = list(ordered_data[:count + 1])
    if len(retval) > count:
        obj = retval[count]
        while '__' in order_field:
            prefix, order_field = order_field.split('__', 1)
            obj = getattr(obj, prefix)
        next = getattr(obj, order_field)
        retval = retval[:-1]
    else:
        next = None

    return (retval, next)

Note that this example includes the "since" object in results (in effect, returning it twice).

[paultag]

👍 adding stock pagination helpers would be a huge help

[paultag]

I ended up doing something like:

class PublicListEndpoint(ListEndpoint):
    methods = ['GET']
    per_page = 100
    serialize_config = {}

    def filter(self, data, **kwargs):
        return data.filter(**kwargs)

    def sort(self, data, sort_by):
        return data.order_by(*sort_by)

    def paginate(self, data, page):
        paginator = Paginator(data, per_page=self.per_page)
        return paginator.page(page)

    def get(self, request, *args, **kwargs):
        params = request.params
        page = 1
        if 'page' in params:
            page = int(params.pop('page'))

        sort_by = []
        if 'sort_by' in params:
            sort_by = params.pop('sort_by').split(",")

        data = self.get_query_set(request, *args, **kwargs)
        data = self.filter(data, **params)
        data = self.sort(data, sort_by)
        try:
            data_page = self.paginate(data, page)
        except EmptyPage:
            raise HttpError(
                404,
                'No such page (heh, literally - its out of bounds)'
            )

        return {
            "meta": {
                "count": len(data_page.object_list),
                "page": page,
                "per_page": self.per_page,
                "max_page": data_page.end_index(),
                "total_count": data.count(),
            },
            "results": [
                serialize(x, **self.serialize_config)
                for x in data_page.object_list
            ]
        }

for an API endpoint I'm hacking on. (just to get a feel for how others are using restless for this)

[senko]

I usually do pagination with offset and limit params (as per the first comment), but I think both are equally valid approaches (IMHO depends on which style you prefer, or if there are specific requirements for the code at hand).

I'm thinking of introducing a ListEndpoint subclass that would allow either usage, and use forms for pagination/filtering/sorting validation. It should be possible to do by overriding serialize, get_query_set, and __init__, something like this :

class OffsetLimitForm(forms.Form):

    offset = forms.IntegerField(required=False, min_value=0)
    limit = forms.IntegerField(required=False, min_value=1)

    def paginate(self, qs):
        meta = None
        if self.is_valid():
            meta = self.cleaned_data.copy()
            meta['count'] = qs.count()
            qs = qs[meta['offset']:meta['offset']+meta['limit']]
        return qs, meta

class PaginatedEndpoint(Endpoint):
    paginate_form = OffsetLimitForm # or PaginateForm

    def __init__(self, *args, **kwargs):
        self.meta = super(MyEndpoint, self).__init__(*args, **kwargs)

    def get_query_set(self, request, *args, **kwargs):
        form = self.paginate_form(request.params)
        qs = super(PaginatedEndpoint, self).get_query_set(request, *args, **kwargs)
        qs, self.meta = form.paginate(qs)

    def serialize(self, objs):
        items = super(PaginatedEndpoint, self).serialize(objs)
        if self.meta is not None:
            return {
                'meta': meta,
                'items': items
            }
        else:
            return items

Then you could just switch forms (or implement your own) to implement the pagination style you need. Analogous for filtering and sorting.

The idea is to add something that wouldn't force you to use a specific way of doing pagination, but would still have the building blocks for the most common forms.