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

def employee_deactivated_list(request, format=None):
    """
    Returns the full employee deactivated list
    ---
    serializer: employees.serializers.EmployeeListSerializer
    responseMessages:
    - code: 401
      message: Unauthorized. Authentication credentials were not provided. Invalid token.
    - code: 403
      message: Forbidden.
    - code: 404
      message: Not found
    """
    if request.method == 'GET':
        employee_list = get_list_or_404(Employee, is_active=False)
        paginator = PageNumberPagination()
        results = paginator.paginate_queryset(employee_list, request)
        serializer = EmployeeListSerializer(results, many=True)
        return paginator.get_paginated_response(serializer.data)

def stars_employee_list(request, employee_id):
    """
    Returns stars list from employee
    ---
    serializer: stars.serializers.StarSerializer
    responseMessages:
    - code: 401
      message: Unauthorized. Authentication credentials were not provided. Invalid token.
    - code: 403
      message: Forbidden, authentication credentials were not provided
    - code: 404
      message: Not found
    """
    if request.method == 'GET':
        employee = get_object_or_404(Employee, pk=employee_id)
        employee_stars = Star.objects.filter(to_user=employee)
        paginator = PageNumberPagination()
        results = paginator.paginate_queryset(employee_stars, request)
        serializer = StarSerializer(results, many=True)
        return paginator.get_paginated_response(serializer.data)

def stars_employee_list_group_by_keyword(request, employee_id):
    """
    Returns stars list from employee grouped by categories
    ---
    serializer: stars.serializers.StarEmployeeKeywordsSerializer
    responseMessages:
    - code: 401
      message: Unauthorized. Authentication credentials were not provided. Invalid token.
    - code: 403
      message: Forbidden, authentication credentials were not provided
    - code: 404
      message: Not found
    """
    if request.method == 'GET':
        employee = get_object_or_404(Employee, pk=employee_id)
        employee_stars = Star.objects.filter(to_user=employee).values(
            'keyword__pk',
            'keyword__name').annotate(num_stars=Count('keyword')).order_by('-num_stars', 'keyword__name')
        paginator = PageNumberPagination()
        result = paginator.paginate_queryset(employee_stars, request)
        serializer = StarEmployeeKeywordsSerializer(result, many=True)
        return paginator.get_paginated_response(serializer.data)

def stars_employee_list_group_by_category_detail(request, employee_id, category_id):
    """
    Returns stars list detail from employee divided by category
    ---
    serializer: stars.serializers.StarSmallSerializer
    responseMessages:
    - code: 401
      message: Unauthorized. Authentication credentials were not provided. Invalid token.
    - code: 403
      message: Forbidden, authentication credentials were not provided
    - code: 404
      message: Not found
    """
    if request.method == 'GET':
        employee = get_object_or_404(Employee, pk=employee_id)
        category = get_object_or_404(Category, pk=category_id)
        stars = Star.objects.filter(to_user=employee, category=category).order_by('-date')
        paginator = PageNumberPagination()
        results = paginator.paginate_queryset(stars, request)
        serializer = StarSmallSerializer(results, many=True)
        return paginator.get_paginated_response(serializer.data)

def stars_employee_list_group_by_keyword_detail(request, employee_id, keyword_id):
    """
    Returns stars list detail from employee divided by keyword
    ---
    serializer: stars.serializers.StarSmallSerializer
    responseMessages:
    - code: 401
      message: Unauthorized. Authentication credentials were not provided. Invalid token.
    - code: 403
      message: Forbidden, authentication credentials were not provided
    - code: 404
      message: Not found
    """
    if request.method == 'GET':
        employee = get_object_or_404(Employee, pk=employee_id)
        keyword = get_object_or_404(Keyword, pk=keyword_id)
        stars = Star.objects.filter(to_user=employee, keyword=keyword).order_by('-date')
        paginator = PageNumberPagination()
        results = paginator.paginate_queryset(stars, request)
        serializer = StarSmallSerializer(results, many=True)
        return paginator.get_paginated_response(serializer.data)

def get_messages(request, employee_id):
    """
    Get all messages for employee id
    ---
    response_serializer: activities.serializers.MessageSerializer
    responseMessages:
    - code: 401
      message: Unauthorized. Authentication credentials were not provided. Invalid token.
    - code: 403
      message: Forbidden.
    - code: 404
      message: Not found
    - code: 500
      message: Internal Server Error
    """
    if request.method == 'GET':
        employee = get_object_or_404(Employee, pk=employee_id)
        messages = Message.objects.filter(
            Q(to_user='all') |
            Q(to_user=employee.location.name) |
            Q(to_user=employee.username))
        paginator = PageNumberPagination()
        results = paginator.paginate_queryset(messages, request)
        serializer = MessageSerializer(results, many=True)
        return paginator.get_paginated_response(serializer.data)

def get_messages_from(request, employee_id):
    """
    Get all messages sent from employee id
    ---
    response_serializer: activities.serializers.MessageSerializer
    responseMessages:
    - code: 401
      message: Unauthorized. Authentication credentials were not provided. Invalid token.
    - code: 403
      message: Forbidden.
    - code: 404
      message: Not found
    - code: 500
      message: Internal Server Error
    """
    if request.method == 'GET':
        employee = get_object_or_404(Employee, pk=employee_id)
        messages = Message.objects.filter(from_user=employee)
        paginator = PageNumberPagination()
        results = paginator.paginate_queryset(messages, request)
        serializer = MessageSerializer(results, many=True)
        return paginator.get_paginated_response(serializer.data)

def get_messages_from_all(request):
    """
    Get all messages sent
    ---
    response_serializer: activities.serializers.MessageSerializer
    responseMessages:
    - code: 401
      message: Unauthorized. Authentication credentials were not provided. Invalid token.
    - code: 403
      message: Forbidden.
    - code: 404
      message: Not found
    - code: 500
      message: Internal Server Error
    """
    if request.method == 'GET':
        messages = Message.objects.all()
        paginator = PageNumberPagination()
        results = paginator.paginate_queryset(messages, request)
        serializer = MessageSerializer(results, many=True)
        return paginator.get_paginated_response(serializer.data)

def my_upcoming_events(request, employee_id):
    """
    Returns the full upcoming events list for employee
    ---
    serializer: events.serializers.EventSerializer
    responseMessages:
    - code: 401
      message: Unauthorized. Authentication credentials were not provided. Invalid token.
    - code: 403
      message: Forbidden.
    - code: 404
      message: Not found
    """
    if request.method == 'GET':
        employee = get_object_or_404(Employee, pk=employee_id)
        events_list = Event.objects.filter(location=employee.location, is_active=True, is_upcoming=True)
        paginator = PageNumberPagination()
        results = paginator.paginate_queryset(events_list, request)
        serializer = EventSerializer(results, many=True)
        return paginator.get_paginated_response(serializer.data)

def local_events(request, employee_id):
    """
    Returns the full upcoming events list for employee location
    ---
    serializer: events.serializers.EventSerializer
    responseMessages:
    - code: 401
      message: Unauthorized. Authentication credentials were not provided. Invalid token.
    - code: 403
      message: Forbidden.
    - code: 404
      message: Not found
    """
    if request.method == 'GET':
        employee = get_object_or_404(Employee, pk=employee_id)
        events = Event.objects.filter(location=employee.location, is_active=True)
        paginator = PageNumberPagination()
        results = paginator.paginate_queryset(events, request)
        serializer = EventSerializer(results, many=True)
        return paginator.get_paginated_response(serializer.data)

def event_activities(request, event_id):
    """
    Returns activity list for event
    ---
    serializer: events.serializers.EventActivitySerializer
    responseMessages:
    - code: 401
      message: Unauthorized. Authentication credentials were not provided. Invalid token.
    - code: 403
      message: Forbidden.
    - code: 404
      message: Not found
    """
    if request.method == 'GET':
        event = get_object_or_404(Event, pk=event_id)
        activities = EventActivity.objects.filter(event=event)
        paginator = PageNumberPagination()
        results = paginator.paginate_queryset(activities, request)
        serializer = EventActivitySerializer(results, many=True)
        return paginator.get_paginated_response(serializer.data)

def test_viewset_factory(self):
        viewset = self.endpoint.get_viewset()
        self.assertEqual(viewset.serializer_class, self.endpoint.get_serializer())

        for backend in (DjangoFilterBackend, filters.SearchFilter):
            self.assertNotIn(backend, viewset.filter_backends)

        self.assertEqual(viewset.__name__, 'ProductViewSet')

        viewset = self.alternate_endpoint.get_viewset()

        for attr in ('permission_classes', 'filter_fields', 'search_fields', 'ordering_fields'):
            self.assertEqual(getattr(viewset, attr), getattr(self, attr))

        for backend in ('DjangoFilterBackend', 'SearchFilter', 'OrderingFilter'):
            self.assertIn(backend, [be.__name__ for be in viewset.filter_backends])

        self.assertEqual(viewset.pagination_class.__name__, 'ProductPagination')
        self.assertTrue(issubclass(
            viewset.pagination_class,
            pagination.PageNumberPagination
        ))

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 _get(self,request,*args,**kwargs):
        """
        ????get() ,?????Name? xxx_class???????????????
        ???? {status,errcode,errmsg,result} ?????????
        ??????APIView??????,???????????ListViewAPI
        """
        # if  request.user.is_authenticated:
        #   return Response({'status':0})  #????????
        ser = SerializerDBTable( self.get_queryset(),many=True)
        data = ser.data
        return Response(data)

        #??????
        page = PageNumberPagination()
        page.paginate_queryset(self.queryset.all()[:200],request)
        return page.get_paginated_response(ser.data)


        return Response({'result':data})

def category_list(request):
    """
    Returns full category list ordered by weight
    ---
    serializer: categories.serializers.CategorySerializer
    parameters:
    - name: pagination
      required: false
      type: string
      paramType: query
    responseMessages:
    - code: 401
      message: Unauthorized. Authentication credentials were not provided. Invalid token.
    - code: 403
      message: Forbidden.
    - code: 404
      message: Not found
    """
    if request.method == 'GET':
        categories = get_list_or_404(Category, is_active=True)
        if request.GET.get('pagination'):
            pagination = request.GET.get('pagination')
            if pagination == 'true':
                paginator = PageNumberPagination()
                results = paginator.paginate_queryset(categories, request)
                serializer = CategorySerializer(results, many=True)
                return paginator.get_paginated_response(serializer.data)
            else:
                return Response(status=status.HTTP_400_BAD_REQUEST)
        else:
            serializer = CategorySerializer(categories, many=True)
            return Response(serializer.data, status=status.HTTP_200_OK)

def keyword_list(request):
    """
    Returns full keyword list ordered by name
    ---
    serializer: categories.serializers.KeywordSerializer
    parameters:
    - name: pagination
      required: false
      type: string
      paramType: query
    responseMessages:
    - code: 401
      message: Unauthorized. Authentication credentials were not provided. Invalid token.
    - code: 403
      message: Forbidden.
    - code: 404
      message: Not found
    """
    if request.method == 'GET':
        keywords = get_list_or_404(Keyword, is_active=True)
        if request.GET.get('pagination'):
            pagination = request.GET.get('pagination')
            if pagination == 'true':
                paginator = PageNumberPagination()
                results = paginator.paginate_queryset(keywords, request)
                serializer = KeywordSerializer(results, many=True)
                return paginator.get_paginated_response(serializer.data)
            else:
                return Response(status=status.HTTP_400_BAD_REQUEST)
        else:
            serializer = KeywordSerializer(keywords, many=True)
            return Response(serializer.data, status=status.HTTP_200_OK)

def stars_keyword_list(request):
    """
    Returns stars list grouped by keyword or result list if you use ?search=
    ---
    serializer: stars.serializers.StarKeywordList
    parameters:
    - name: search
      required: false
      type: string
      paramType: query
    responseMessages:
    - code: 401
      message: Unauthorized. Authentication credentials were not provided. Invalid token.
    - code: 403
      message: Forbidden, authentication credentials were not provided
    - code: 404
      message: Not found
    """
    if request.method == 'GET':
        if request.GET.get('search'):
            search_term = request.GET.get('search')
            star_list = Star.objects.filter(
                Q(keyword__name__icontains=search_term)).values(
                    'keyword__pk',
                    'keyword__name').annotate(num_stars=Count('keyword')).order_by('-num_stars')
        else:
            star_list = Star.objects.all().values(
                'keyword__pk',
                'keyword__name').annotate(num_stars=Count('keyword')).order_by('-num_stars')
        paginator = PageNumberPagination()
        results = paginator.paginate_queryset(star_list, request)
        serializer = StarKeywordList(results, many=True)
        return paginator.get_paginated_response(serializer.data)

def stars_keyword_list_detail(request, keyword_id):
    """
    Returns stars list detail for keyword id.
    ---
    response_serializer: stars.serializers.StarTopEmployeeLists
    responseMessages:
    - code: 401
      message: Unauthorized. Authentication credentials were not provided. Invalid token.
    - code: 403
      message: Forbidden, authentication credentials were not provided
    - code: 404
      message: Not found
    """
    if request.method == 'GET':
        keyword = get_object_or_404(Keyword, pk=keyword_id)
        stars = Star.objects.filter(keyword=keyword).values(
            'to_user__pk',
            'to_user__username',
            'to_user__first_name',
            'to_user__last_name',
            'to_user__level',
            'to_user__avatar').annotate(num_stars=Count('keyword')).order_by('-num_stars')
        paginator = PageNumberPagination()
        results = paginator.paginate_queryset(stars, request)
        serializer = StarTopEmployeeLists(results, many=True)
        return paginator.get_paginated_response(serializer.data)

def employee_list_group_by_badges(request):
    """
    Returns badge list with employee counter or result list if you use ?search=
    ---
    response_serializer: stars.serializers.EmployeeBadgeListSerializer
    parameters:
    - name: search
      required: false
      type: string
      paramType: query
    responseMessages:
    - code: 401
      message: Unauthorized. Authentication credentials were not provided. Invalid token.
    - code: 403
      message: Forbidden, authentication credentials were not provided
    - code: 404
      message: Not found
    """
    if request.method == 'GET':
        if request.GET.get('search'):
            search_term = request.GET.get('search')
            badge_list = EmployeeBadge.objects.filter(
                Q(badge__name__icontains=search_term)).values(
                    'badge__pk',
                    'badge__name').annotate(num_employees=Count('to_user')).order_by('-num_employees')
        else:
            badge_list = EmployeeBadge.objects.all().values(
                'badge__pk',
                'badge__name').annotate(num_employees=Count('to_user')).order_by('-num_employees')
        paginator = PageNumberPagination()
        results = paginator.paginate_queryset(badge_list, request)
        serializer = EmployeeBadgeListSerializer(results, many=True)
        return paginator.get_paginated_response(serializer.data)

def employee_list_group_by_badges_detail(request, badge_id):
    """
    Returns employee list grouped by badge, you should provide badge_id
    ---
    response_serializer: stars.serializers.EmployeeGroupedListSerializer
    responseMessages:
    - code: 401
      message: Unauthorized. Authentication credentials were not provided. Invalid token.
    - code: 403
      message: Forbidden, authentication credentials were not provided
    - code: 404
      message: Not found
    """
    if request.method == 'GET':
        badge = get_object_or_404(Badge, pk=badge_id)
        employee_list = EmployeeBadge.objects.filter(badge=badge).values(
            'to_user__pk',
            'to_user__username',
            'to_user__first_name',
            'to_user__last_name',
            'to_user__level',
            'to_user__avatar')
        paginator = PageNumberPagination()
        results = paginator.paginate_queryset(employee_list, request)
        serializer = EmployeeGroupedListSerializer(results, many=True)
        return paginator.get_paginated_response(serializer.data)

def get_notifications(request, employee_id):
    """
    Get all notifications for employee id
    ---
    response_serializer: activities.serializers.NotificationSerializer
    responseMessages:
    - code: 401
      message: Unauthorized. Authentication credentials were not provided. Invalid token.
    - code: 403
      message: Forbidden.
    - code: 404
      message: Not found
    - code: 500
      message: Internal Server Error
    """
    if request.method == 'GET':
        employee = get_object_or_404(Employee, pk=employee_id)
        activities = Activity.objects.annotate(
            profile=F('to_user')).values('datetime',
                                         'text',
                                         'profile').filter(to_user=employee)
        messages = Message.objects.annotate(
            profile=F('from_user')).values('datetime',
                                           'text',
                                           'profile').filter(Q(to_user='all') |
                                                             Q(to_user=employee.location.name) |
                                                             Q(to_user=employee.username))
        notifications = list(chain(activities, messages))
        notifications = sorted(notifications, reverse=True)
        paginator = PageNumberPagination()
        results = paginator.paginate_queryset(notifications, request)
        serializer = NotificationSerializer(results, many=True)
        return paginator.get_paginated_response(serializer.data)

def other_location_events(request, employee_id):
    """
    Returns the full upcoming events list for employee location
    ---
    serializer: events.serializers.EventSerializer
    responseMessages:
    - code: 401
      message: Unauthorized. Authentication credentials were not provided. Invalid token.
    - code: 403
      message: Forbidden.
    - code: 404
      message: Not found
    """
    events = []
    if request.method == 'GET':
        employee = get_object_or_404(Employee, pk=employee_id)
        events_list = Event.objects.filter(is_active=True)

        for event in events_list:
            if event.location != employee.location:
                events.append(event)

        paginator = PageNumberPagination()
        results = paginator.paginate_queryset(events, request)
        serializer = EventSerializer(results, many=True)
        return paginator.get_paginated_response(serializer.data)

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

        :param args:
        :param kwargs:
        """
        self.facets = None
        # self.page = None
        # self.request = None
        super(PageNumberPagination, self).__init__(*args, **kwargs)

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 paginated_queryset(queryset, request, pagination_class=PageNumberPagination()):
    '''
        Return a paginated result for a queryset
    '''
    paginator = pagination_class
    paginator.page_size = settings.REST_FRAMEWORK['PAGE_SIZE']
    result_page = paginator.paginate_queryset(queryset, request)
    return (paginator, result_page)

def employee_list(request, format=None):
    """
    Returns the full employee list or result list if you use ?search=
    ---
    serializer: employees.serializers.EmployeeListSerializer
    parameters:
    - name: search
      required: false
      type: string
      paramType: query
    responseMessages:
    - code: 401
      message: Unauthorized. Authentication credentials were not provided. Invalid token.
    - code: 403
      message: Forbidden.
    - code: 404
      message: Not found
    """
    if request.method == 'GET':
        if request.GET.get('search'):
            request_terms = request.GET.get('search')
            search_terms_array = request_terms.split()

            initial_term = search_terms_array[0]
            employee_list = Employee.objects.filter(
                Q(first_name__icontains=initial_term) |
                Q(last_name__icontains=initial_term) |
                Q(username__icontains=initial_term)).filter(is_active=True, is_base_profile_complete=True)
            if len(search_terms_array) > 1:
                for term in range(1, len(search_terms_array)):
                    employee_list = employee_list.filter(
                        Q(first_name__icontains=search_terms_array[term]) |
                        Q(last_name__icontains=search_terms_array[term]) |
                        Q(username__icontains=search_terms_array[term])).filter(
                                                                            is_active=True,
                                                                            is_base_profile_complete=True)
        else:
            employee_list = get_list_or_404(Employee, is_active=True, is_base_profile_complete=True)
        paginator = PageNumberPagination()
        results = paginator.paginate_queryset(employee_list, request)
        serializer = EmployeeListSerializer(results, many=True)
        return paginator.get_paginated_response(serializer.data)

def list(self, request):  # pylint: disable=too-many-locals

        query_objects = self.class_name.objects  # pylint: disable=no-member
        queryset = query_objects.all().order_by(
            self.order_param_name
            )
        if request.GET.get('search'):
            queryset = []
            param = self.request.query_params.get('search', None)
            if param is not None:
                fields_class = self.class_name
                fields = fields_class._meta.fields  # pylint: disable=no-member

                fields_char = [
                    f for f in fields
                    if isinstance(f, CharField)]
                fields = [
                    f for f in fields
                    if not isinstance(f, CharField)]
                queries_char = [
                    Q(**{f.name+'__contains': param})
                    for f in fields_char]
                try:
                    param_num = int(param)
                except ValueError:
                    param_num = 0
                queries = [Q(**{f.name: param_num}) for f in fields]
                print(fields)
                _queries = Q()
                for query in queries:
                    _queries = _queries | query
                for query in queries_char:
                    _queries = _queries | query
                objects = self.class_name.objects  # pylint: disable=no-member
                queryset = objects.filter(_queries)
        if request.GET.get('all'):
            self.pagination_class = None
            serializer = self.serializer_class(  # pylint: disable=not-callable
                queryset,
                many=True)
            response = Response(serializer.data)
        else:
            paginator = pagination.PageNumberPagination()
            queryset = paginator.paginate_queryset(
                queryset=queryset,
                request=request
                )
            serializer = self.serializer_class(  # pylint: disable=not-callable
                queryset,
                many=True)
            response = paginator.get_paginated_response(serializer.data)

        return response