使用Django Rest Framework CSV和StreamingHttpResponse

Question

我有一个标准的DRF Web应用程序，其中一个路由输出CSV数据。渲染整个CSV表示需要一些时间。数据集非常大，因此我想使用流式HTTP响应，以便客户端不会超时。但是，使用https://github.com/mjumbewu/django-rest-framework-csv/blob/2ff49cff4b81827f3f450fd7d56827c9671c5140/rest_framework_csv/renderers.py#L197提供的示例并不能完全实现这一点。数据仍然是一个大负载，而不是分块的，客户端最终等待接收到字节才会响应。结构类似于以下内容：

models.py

class Report(models.Model):
  count = models.PostiveIntegerField(blank=True)
  ...

renderers.py

class ReportCSVRenderer(CSVStreamingRenderer):
  header = ['count']

serializers.py

class ReportSerializer(serializers.ModelSerializer):
  count = fields.IntegerField()

  class Meta:
    model = Report

views.py

class ReportCSVView(generics.Viewset, mixins.ListModelMixin):
  def get_queryset(self):
    return Report.objects.all()

  def list(self, request, *args, **kwargs):
    queryset = self.get_queryset()
    data = ReportSerializer(queryset, many=True)
    renderer = ReportCSVRenderer()

    response = StreamingHttpResponse(renderer.render(data), content_type='text/csv')
    response['Content-Disposition'] = 'attachment; filename="f.csv"'

    return response

注意：必须注释或更改一些内容。

谢谢

- 3066d0

Answer 1

受 @3066d0 的启发，有一个更简单的解决方案：

renderers.py

class ReportsRenderer(CSVStreamingRenderer):
    header = [ ... ]
    labels = { ... }

views.py

class ReportCSVViewset(ListModelMixin, GenericViewSet):
    queryset = Report.objects.select_related('stuff')
    serializer_class = ReportCSVSerializer
    renderer_classes = [ReportsRenderer]
    PAGE_SIZE = 1000

    def list(self, request, *args, **kwargs):
        queryset = self.filter_queryset(self.get_queryset())
        response = StreamingHttpResponse(
            request.accepted_renderer.render(self._stream_serialized_data(queryset)),
            status=200,
            content_type="text/csv",
        )
        response["Content-Disposition"] = 'attachment; filename="reports.csv"'
        return response

    def _stream_serialized_data(self, queryset):
        serializer = self.get_serializer_class()
        paginator = Paginator(queryset, self.PAGE_SIZE)
        for page in paginator.page_range:
            yield from serializer(paginator.page(page).object_list, many=True).data

关键是你需要传递一个生成器作为渲染器的 data 参数，它会生成序列化的数据，然后 CSVStreamingRenderer 会自己完成操作并流式传输响应。我更喜欢这种方法，因为这样你不需要重写第三方库的代码。

- Andrii Vityk

Answer 2

Django的StreamingHttpResponse对于小响应比传统的HttpResponse要慢得多。

如果不需要，请勿使用它；实际上，Django文档建议只在绝对需要在将数据传输到客户端之前不迭代整个内容时才使用StreamingHttpResponse。

此外，对于您的问题，您可能会发现设置chunk_size、切换到FileResponse或返回普通的Response（如果使用REST框架）或HttpResponse很有用。

编辑1：关于设置块大小：

在文件api中，您可以分块打开文件，因此并非所有文件都会加载到内存中。

希望这对您有所帮助。

- ascoder

---

对于你的问题，设置chunk_size可能会有所帮助。那么在哪里可以设置chunk_size呢？ - Will Keeling

在 File api 中，您可以分块打开文件，以便不会将整个文件加载到内存中。 - ascoder

谢谢你的帮助！ :) - 3066d0

v4.2 版本的文件文档已更新：https://docs.djangoproject.com/en/4.2/ref/files/file/ - undefined

Answer 3

在渲染数据时，您需要提供CSV标题（通过header参数）：

renderer.render(data, renderer_context={'header': ['header1', 'header2', 'header3']})

如果您不指定 header 参数，djangorestframework-csv 将尝试自行“猜测” CSV 标头。为了“猜测”CSV标头，djangorestframework-csv会将所有 data 加载到内存中，这会导致您遇到的延迟问题。

- spg

Answer 4

所以最终我使用带有查询集的 Paginator 类找到了一个让我满意的解决方案。首先，我编写了一个继承自 CSVStreamingRenderer 的渲染器，然后在我的 CSVViewset 的渲染器中使用它。

renderers.py

from rest_framework_csv.renderers import CSVStreamingRenderer

# *****************************************************************************
# BatchedCSVRenderer
# *****************************************************************************


class BatchedCSVRenderer(CSVStreamingRenderer):

    """
    a CSV renderer that works with large querysets returning a generator
    function. Used with a streaming HTTP response, it provides response bytes
    instead of the client waiting for a long period of time
    """

    def render(self, data, renderer_context={}, *args, **kwargs):
        if 'queryset' not in data:
            return data

        csv_buffer = Echo()
        csv_writer = csv.writer(csv_buffer)

        queryset = data['queryset']
        serializer = data['serializer']

        paginator = Paginator(queryset, 50)

        #  rendering the header or label field was taken from the tablize
        #  method in django rest framework csv

        header = renderer_context.get('header', self.header)
        labels = renderer_context.get('labels', self.labels)

        if labels:
            yield csv_writer.writerow([labels.get(x, x) for x in header])
        else:
            yield csv_writer.writerow(header)

        for page in paginator.page_range:
            serialized = serializer(
                paginator.page(page).object_list, many=True
            ).data

            #  we use the tablize function on the parent class to get a
            #  generator that we can use to yield a row

            table = self.tablize(
                serialized,
                header=header,
                labels=labels,
            )

            #  we want to remove the header from the tablized data so we use
            #  islice to take from 1 to the end of generator

            for row in itertools.islice(table, 1, None):
                yield csv_writer.writerow(row)

# *****************************************************************************
# ReportsRenderer
# *****************************************************************************


class ReportsRenderer(BatchedCSVRenderer):

    """
    A render for returning CSV data for reports

    """

    header = [ ... ]
    labels = { ... }

views.py

from django.http import StreamingHttpResponse
from rest_framework import mixins, viewsets

# *****************************************************************************
# CSVViewSet
# *****************************************************************************


class CSVViewSet(
        mixins.ListModelMixin,
        viewsets.GenericViewSet,
):

    def list(self, request, *args, **kwargs):
        queryset = self.get_queryset()

        return StreamingHttpResponse(
            request.accepted_renderer.render({
                'queryset': queryset,
                'serializer': self.get_serializer_class(),
            })
)

# *****************************************************************************
# ReportsViewset
# *****************************************************************************


class ReportCSVViewset(CSVViewSet):

    """
    Viewset for report CSV output

    """

    renderer_classes = [ReportCSVRenderer]
    serializer_class = serializers.ReportCSVSerializer

    def get_queryset(self):
        queryset = Report.objects.filter(...)

这可能看起来对于流式响应来说有点多，但我们在其他许多地方使用了BatchedCSVRender和CSVViewset。如果您将服务器运行在nginx后面，则调整那里的设置以允许流式响应可能也很有用。

希望这能帮助有同样目标的任何人。请告诉我是否还有其他信息可以提供。

- 3066d0