DRF: DataFrame を使って CSV エクスポート

CSV レンダラー

def dataframe_to_csv_stream(df, encoding=None, **kwargs):
    buffer = BytesIO()
    df.to_csv(buffer, **kwargs)
    buffer.seek(0)
    if encoding:
        return BytesIO(buffer.read().decode("utf8").encode(encoding))
    return buffer

class DataFrameCSVRenderer(BaseRenderer):
    media_type = "text/csv"
    format = "csv"

    def render(self, data, media_type=None, renderer_context=None, writer_opts=None):
        renderer_context = renderer_context or {}
        if not isinstance(data, pd.DataFrame):
            data = pd.DataFrame(data)
        stream = dataframe_to_csv_stream(data, index=False)
        return stream.read().decode("utf-8-sig")

DataFrame を返すリストシリアライザ

class DataFrameSerializer(serializers.ListSerializer):
    @property
    def data(self):
        ret = super().data
        ser = ret.serializer.child
        columns = dict((name, field.label) for name, field in ser.fields.items())
        return pd.DataFrame(ret).rename(columns=columns)

エクスポートするシリアライザを生成するメタクラス

class ExpporSerializerMetaclass(serializers.SerializerMetaclass):
    EXPORT_META = {}
    FIELD_CLASS_DEFAULTS = {}

    @classmethod
    def create_fields(cls):
        def _create_field(item):
            label, name, klass = item
            defaults = cls.FIELD_CLASS_DEFAULTS[klass]
            return (name, getattr(serializers, klass)(label=label, **defaults))

        return dict(map(_create_field, cls.EXPORT_META))

    @classmethod
    def annotate(cls, queryset):
        raise NotImplementedError()

    def __new__(cls, name, bases, attrs, **kwargs):
        attrs["Meta"].fields = [i[1] for i in cls.EXPORT_META]      # CSVで返答するカラム
        attrs["Meta"].list_serializer_class = DataFrameSerializer   # リストシリアライザ
        attrs.update(**cls.create_fields())
        attrs["annotate"] = cls.annotate    # クエリセットをアノテートさせるクラスメソッド 

        return super().__new__(cls, name, bases, attrs, **kwargs)

実際のメタクラス

class IncomeExpporSerializerMetaclass(ExpporSerializerMetaclass):
    # (ラベル, フィールド名, フィールドクラス名)
    EXPORT_META = [
        ("会社コード", "order__company_code", "CharField"),
        ....
    ]

    # フィールドクラスのデフォルト値
    FIELD_CLASS_DEFAULTS = dict(
        CharField=dict(read_only=True),
        DecimalField=dict(max_digits=12, decimal_places=1, read_only=True),
    )

    @classmethod
    def annotate(cls, queryset):
        """ クエリセットを適切にアノテートする"""
        def _ann(item):
            label, name, klass = item
            if name.startswith("order__"):
                # リレーション
                return (name, F(name))
            if name.startswith("meme__"):
                # 固定値
                return (name, Value(""))
            # その他はモデルフィールドなのでアノテートしない
            return (None, None)

        ann = dict(filter(lambda i: i[1] is not None, map(_ann, cls.EXPORT_META)))
        return queryset.annotate(**ann)

実際のシリアライザ

class IncomeExportSerializer(serializers.ModelSerializer, metaclass=IncomeExpporSerializerMetaclass):
    class Meta:
        model = models.Income

API 本体(ビューセット)

class IncomeViewSet(VS.BaseModelViewSet):

    serializer_class = serializers.IncomeSerializer
    queryset = models.Income.objects.all()
    pagination_class = paginations.Pagination

    renderer_classes = tuple(api_settings.DEFAULT_RENDERER_CLASSES) + (DataFrameCSVRenderer,)

    def get_serializer_class(self):
        """ export アクションでシリアライザを切り替える """
        res = {
            "export": serializers.IncomeExportSerializer,
        }.get(self.action, super().get_serializer_class())
        return res

    def get_queryset(self):
        """ export アクションの場合、必要なアノテーションを行う """
        qs = super().get_queryset()
        if self.action == "export":
            qs = serializers.IncomeExportSerializer.annotate(qs).filter(level=0)
            return qs
        return qs

    @decorators.action(methods=["get"], detail=False)
    def export(self, request, pk=None):
        ## リスト処理を行うと シリアライザとレンダラで勝手にCSVを返答する(Accept: text/csvの時)
        return self.list(request)