如何在Django Rest Framework中为Google BigQuery返回数据添加分页

问题背景 #

从Google BigQuery获取现金流数据后，在Django Rest Framework的CashFlowList视图中需要添加分页功能，且未使用Serializer处理输出。当前视图直接构造Response返回数据，无法生成包含count、previous、next字段的期望格式。

当前视图代码 #

class CashFlowList(generics.ListAPIView):
    permission_classes = [permissions.IsAuthenticated, TokenHasReadWriteScope]
    pagination_class = PageNumberPagination

    @staticmethod
    def cash_flow_list(self, rows, info, *args, **kwargs):
        data = {'invoices': rows}
        message = {'info': info, 'data': data}
        return message

    def get(self, request):
    
        try:
            dataset = 'dummy_dataset'
            get_cashflow_query = f" SELECT  ref_no AS invoice_number, party_name AS customer_name,credit_amount, pending_amount," \
                             f" DATE_ADD(date, INTERVAL credit_period DAY) AS due_date, DATE_DIFF(CURRENT_TIMESTAMP(), " \
                             f" DATE_ADD(date, INTERVAL credit_period DAY) , DAY) AS due_days FROM {dataset}.{CASH_FLOW_TABLE}"

            cashflow_list_data = cashflow_summary_obj.get_cash_values(get_cashflow_query)
            data = []
            for row in cashflow_list_data:
                data.append({'invoice_number':row[0], 'customer_name': row[1], 'credit_amount': row[2], 
                        'pending_amount': row[3], 'due_date': row[4], 'due_days': row[5]})
            info = {'status_code': '131', 'status': 'SUCCESS', 'message': "cash flow data list."}
            message = self.cash_flow_list(self, info=info, rows=data)
        except NotFound:
            info = {'status_code': '711', 'status': 'FAIL',
                'message': 'Unknown dataset, Please call Administrator...!'}
            data = {'total_receivables': 0, 'total_pending_receivables': 0, 'customers_overdue': 0}
            message = self.cash_flow_list(self, info=info, rows=data)
        return Response(data=message, status=status.HTTP_200_OK)

当前输出格式 #

{
  "info": {
    "status_code": "131",
    "status": "SUCCESS",
    "message": "cash flow data list."
  },
  "data": {
    "invoices": [
      {
        "invoice_number": 3,
        "customer_name": "Philip",
        "credit_amount": 25000,
        "pending_amount": 7760,
        "due_date": "2022-12-15T00:00:00Z",
        "due_days": 33
      }
      // 其他数据项省略
    ]
  }
}

期望输出格式 #

{
  "info": {
    "status_code": "131",
    "status": "SUCCESS",
    "message": "cash flow data list."
  },
  "data": {
    "count":5,
    "previous": null,
    "next": null,
    "invoices": [
      {
        "invoice_number": 3,
        "customer_name": "Philip",
        "credit_amount": 25000,
        "pending_amount": 7760,
        "due_date": "2022-12-15T00:00:00Z",
        "due_days": 33
      }
      // 其他数据项省略
    ]
  }
}

解决方案 #

原代码未触发DRF分页逻辑的核心原因是重写了get方法并直接返回自定义Response，绕过了ListAPIView的分页处理流程。以下两种方式可实现需求：

方式一：自定义分页类并手动调用 #

1. 自定义分页类，适配期望的data结构

from rest_framework.pagination import PageNumberPagination

class CashFlowPagination(PageNumberPagination):
    page_size = 10  # 自定义每页默认数量
    page_size_query_param = 'page_size'  # 允许通过请求参数调整每页数量
    max_page_size = 100  # 限制最大每页数量

    def get_paginated_response(self, data):
        # 重写返回结构，替换默认的results字段为invoices
        return {
            'count': self.page.paginator.count,
            'previous': self.get_previous_link(),
            'next': self.get_next_link(),
            'invoices': data
        }

2. 修改视图，调用分页器处理数据并构造最终响应

class CashFlowList(generics.ListAPIView):
    permission_classes = [permissions.IsAuthenticated, TokenHasReadWriteScope]
    pagination_class = CashFlowPagination

    def get(self, request):
        try:
            dataset = 'dummy_dataset'
            get_cashflow_query = f"SELECT ref_no AS invoice_number, party_name AS customer_name, credit_amount, pending_amount," \
                                 f" DATE_ADD(date, INTERVAL credit_period DAY) AS due_date, DATE_DIFF(CURRENT_TIMESTAMP(), " \
                                 f" DATE_ADD(date, INTERVAL credit_period DAY), DAY) AS due_days FROM {dataset}.{CASH_FLOW_TABLE}"

            cashflow_list_data = cashflow_summary_obj.get_cash_values(get_cashflow_query)
            invoices_data = []
            for row in cashflow_list_data:
                invoices_data.append({
                    'invoice_number': row[0],
                    'customer_name': row[1],
                    'credit_amount': row[2],
                    'pending_amount': row[3],
                    'due_date': row[4],
                    'due_days': row[5]
                })
            
            # 初始化分页器并处理数据
            paginator = self.pagination_class()
            paginated_invoices = paginator.paginate_queryset(invoices_data, request)
            paginated_data = paginator.get_paginated_response(paginated_invoices).data
            
            # 构造最终响应结构
            info = {'status_code': '131', 'status': 'SUCCESS', 'message': "cash flow data list."}
            final_response = {'info': info, 'data': paginated_data}
            
            return paginator.get_paginated_response(final_response)
        except NotFound:
            info = {'status_code': '711', 'status': 'FAIL',
                    'message': 'Unknown dataset, Please call Administrator...!'}
            # 异常场景返回符合格式的空数据
            error_data = {
                'count': 0,
                'previous': None,
                'next': None,
                'invoices': []
            }
            final_response = {'info': info, 'data': error_data}
            return Response(final_response, status=status.HTTP_200_OK)

方式二：手动实现分页逻辑 #

如果不想依赖DRF的分页类，可以手动处理分页参数、数据切片和链接生成：

class CashFlowList(generics.ListAPIView):
    permission_classes = [permissions.IsAuthenticated, TokenHasReadWriteScope]
    # 定义分页参数
    PAGE_SIZE = 10

    def get(self, request):
        try:
            dataset = 'dummy_dataset'
            get_cashflow_query = f"SELECT ref_no AS invoice_number, party_name AS customer_name, credit_amount, pending_amount," \
                                 f" DATE_ADD(date, INTERVAL credit_period DAY) AS due_date, DATE_DIFF(CURRENT_TIMESTAMP(), " \
                                 f" DATE_ADD(date, INTERVAL credit_period DAY), DAY) AS due_days FROM {dataset}.{CASH_FLOW_TABLE}"

            cashflow_list_data = cashflow_summary_obj.get_cash_values(get_cashflow_query)
            invoices_data = []
            for row in cashflow_list_data:
                invoices_data.append({
                    'invoice_number': row[0],
                    'customer_name': row[1],
                    'credit_amount': row[2],
                    'pending_amount': row[3],
                    'due_date': row[4],
                    'due_days': row[5]
                })
            
            # 解析分页参数
            page = int(request.query_params.get('page', 1))
            page_size = int(request.query_params.get('page_size', self.PAGE_SIZE))
            page_size = min(page_size, 100)  # 限制最大每页数量
            
            # 计算切片范围
            total_count = len(invoices_data)
            start_idx = (page - 1) * page_size
            end_idx = start_idx + page_size
            paginated_invoices = invoices_data[start_idx:end_idx]
            
            # 生成上下页链接
            base_url = request.build_absolute_uri().split('?')[0]
            previous_link = f"{base_url}?page={page-1}&page_size={page_size}" if page > 1 else None
            next_link = f"{base_url}?page={page+1}&page_size={page_size}" if end_idx < total_count else None
            
            # 构造分页后的数据结构
            paginated_data = {
                'count': total_count,
                'previous': previous_link,
                'next': next_link,
                'invoices': paginated_invoices
            }
            
            info = {'status_code': '131', 'status': 'SUCCESS', 'message': "cash flow data list."}
            final_response = {'info': info, 'data': paginated_data}
            return Response(final_response, status=status.HTTP_200_OK)
        except NotFound:
            info = {'status_code': '711', 'status': 'FAIL',
                    'message': 'Unknown dataset, Please call Administrator...!'}
            error_data = {
                'count': 0,
                'previous': None,
                'next': None,
                'invoices': []
            }
            final_response = {'info': info, 'data': error_data}
            return Response(final_response, status=status.HTTP_200_OK)

内容的提问来源于stack exchange，提问作者Jin