Python rest_framework.pagination 模块，LimitOffsetPagination() 实例源码

def list(request, user_pk=None) -> Response:
        """List transactions for a single user

        :param request: Request send from the client
        :param user_pk: Primary key to identify a user
        :return: Response
        """
        try:
            user = User.objects.get(id=user_pk)
        except User.DoesNotExist:
            return Response(data={'msg': 'user {} not found'.format(user_pk)}, status=status.HTTP_404_NOT_FOUND)
        paginator = LimitOffsetPagination()
        paginator.max_limit = 250
        paginator.default_limit = 100
        transactions = paginator.paginate_queryset(Transaction.objects.filter(user=user), request)
        return Response(data={'entries': [x.to_dict() for x in transactions], 'limit': paginator.limit,
                              'offset': paginator.offset, 'overall_count': paginator.count},
                        status=status.HTTP_200_OK)

def get_paginator_serializer(self, view, child_serializer_class):
        class BaseFakeListSerializer(serializers.Serializer):
            results = child_serializer_class(many=True)

        class FakePrevNextListSerializer(BaseFakeListSerializer):
            next = URLField()
            previous = URLField()

        # Validate if the view has a pagination_class
        if not (hasattr(view, 'pagination_class')) or view.pagination_class is None:
            return BaseFakeListSerializer

        pager = view.pagination_class
        if hasattr(pager, 'default_pager'):
            # Must be a ProxyPagination
            pager = pager.default_pager

        if issubclass(pager, (PageNumberPagination, LimitOffsetPagination)):
            class FakeListSerializer(FakePrevNextListSerializer):
                count = IntegerField()
            return FakeListSerializer
        elif issubclass(pager, CursorPagination):
            return FakePrevNextListSerializer

        return BaseFakeListSerializer

def get_pagination_introspector(view, si=None):
    """
    Create pagination introspector based on view
    :param view: DjangoRestFramework view
    :param si: SerializerIntrospector
    :return: PaginationIntrospector
    :rtype: BasePaginationIntrospector
    """
    if getattr(view, 'pagination_class', None):
        # DjangoRestFramework 3.0 pagination style with pagination class
        pagination_class = view.pagination_class
        from rest_framework import pagination
        if pagination_class == pagination.PageNumberPagination:
            return PageNumberPaginationIntrospector(view, pagination_class, si=si)
        elif pagination_class == pagination.LimitOffsetPagination:
            return LimitOffsetPaginationIntrospector(view, pagination_class, si=si)
        elif pagination_class == pagination.CursorPagination:
            return CursorPaginationIntrospector(view, pagination_class, si=si)
        else:
            return BasePaginationIntrospector(view, pagination_class, si=si)
    else:
        # Unrecognized view type
        return BasePaginationIntrospector(si=si)

def __init__(self, *args, **kwargs):
        """Constructor.

        :param args:
        :param kwargs:
        """
        self.facets = None
        # self.count = None
        # self.limit = None
        # self.offset = None
        # self.request = None
        super(LimitOffsetPagination, self).__init__(*args, **kwargs)

def get(self, request, *args, **kwargs):
        query = self.get_query()
        sort = self.get_sort()

        pagination = LimitOffsetPagination()
        pagination.default_limit = 20
        pagination.limit = pagination.get_limit(request)
        pagination.offset = pagination.get_offset(request)
        pagination.request = request

        body = {
            'query': query,
            'size': pagination.limit,
            'from': pagination.offset
        }
        if sort:
            body['sort'] = sort

        # execute elasticsearch query
        indexer = self.get_indexer()
        res = indexer.search(body=body)

        # map back to expected format
        items = list(indexer.map_results(res['hits']['hits']))
        pagination.count = res['hits']['total']
        return pagination.get_paginated_response(items)

def list(request) -> Response:
        """List users

        :param request: HTTP Request
        :return: Response
        """
        paginator = LimitOffsetPagination()
        paginator.max_limit = 250
        paginator.default_limit = 100
        users = paginator.paginate_queryset(User.objects.filter(active=True), request)
        return Response(
            data={'entries': [x.to_dict() for x in users], 'limit': paginator.limit,
                  'offset': paginator.offset, 'overall_count': paginator.count},
            status=status.HTTP_200_OK)

def list(request):
        """List transactions for all users

        :param request: Request send from the client
        :return: Response
        """
        paginator = LimitOffsetPagination()
        paginator.max_limit = 250
        paginator.default_limit = 100
        transactions = paginator.paginate_queryset(Transaction.objects.all(), request)
        return Response(data={'entries': [x.to_dict() for x in transactions], 'limit': paginator.limit,
                              'offset': paginator.offset, 'overall_count': paginator.count},
                        status=status.HTTP_200_OK)

def pagination_factory(endpoint):
    pg_cls_name = '{}Pagination'.format(endpoint.model.__name__)

    page_size = getattr(endpoint, 'page_size', None)
    pg_cls_attrs = {
        'page_size': page_size if page_size is not None else settings.REST_FRAMEWORK.get('PAGE_SIZE', 50),
    }

    if hasattr(endpoint, 'pagination_template'):
        pg_cls_attrs['template'] = endpoint.pagination_template

    BasePagination = getattr(endpoint, 'base_pagination_class', pagination.PageNumberPagination)
    if issubclass(BasePagination, pagination.PageNumberPagination):
        pg_cls_attrs['page_size_query_param'] = getattr(endpoint, 'page_size_query_param', 'page_size')
        for param in ('django_paginator_class', 'page_query_param', 'max_page_size', 'last_page_string',
                      'page_size'):
            if getattr(endpoint, param, None) is not None:
                pg_cls_attrs[param] = getattr(endpoint, param)
    elif issubclass(BasePagination, pagination.LimitOffsetPagination):
        pg_cls_attrs.pop('page_size')
        for param in ('default_limit', 'limit_query_param', 'offset_query_param', 'max_limit'):
            if getattr(endpoint, param, None) is not None:
                pg_cls_attrs[param] = getattr(endpoint, param)
    elif issubclass(BasePagination, pagination.CursorPagination):
        for param in ('page_size', 'cursor_query_param', 'ordering'):
            if getattr(endpoint, param, None) is not None:
                pg_cls_attrs[param] = getattr(endpoint, param)
    else:
        raise ImproperlyConfigured('base_pagination_class needs to be a subclass of one of the following:'
                                   'PageNumberPagination, LimitOffsetPagination, CursorPagination')

    return type(pg_cls_name, (BasePagination, ), pg_cls_attrs)

def paginate_queryset(self, queryset, request, view=None):
        if 'offset' in request.GET or 'limit' in request.GET:
            self.proxy = LimitOffsetPagination
        return self.proxy.paginate_queryset(self, queryset, request, view)

def get(self, request):
        """GET request for the list the history of purchases of the shop."""
        purchases = Purchase.objects.annotate(num_orders=Count('orders'))\
                                    .filter(num_orders__gt=0)
        paginator = LimitOffsetPagination()
        result_page = paginator.paginate_queryset(purchases, request)
        return Response(self.serializer_class(result_page, many=True).data)