ref: replace CsvMixin with a typesafe alternative (#74967)

mypy 1.11 points out that the approach here is unsafe

CsvMixin is used in getsentry so I'll follow up with deleting it after
converting the getsentry usages over as well
<!-- Describe your PR here. -->

pyproject.toml

@@ -617,6 +617,7 @@ module = [
     "sentry.utils.urls",
     "sentry.utils.uwsgi",
     "sentry.utils.zip",
+    "sentry.web.frontend.csv",
     "sentry_plugins.base",
     "tests.sentry.api.endpoints.issues.*",
     "tests.sentry.event_manager.test_event_manager",

src/sentry/web/frontend/csv.py

@@ -0,0 +1,40 @@
+from __future__ import annotations
+
+import csv
+from collections.abc import Generator, Iterable
+from typing import Generic, TypeVar
+
+from django.http import StreamingHttpResponse
+
+T = TypeVar("T")
+
+
+# csv.writer doesn't provide a non-file interface
+# https://docs.djangoproject.com/en/1.9/howto/outputting-csv/#streaming-large-csv-files
+class Echo:
+    def write(self, value: str) -> str:
+        return value
+
+
+class CsvResponder(Generic[T]):
+    def get_header(self) -> tuple[str, ...]:
+        raise NotImplementedError
+
+    def get_row(self, item: T) -> tuple[str, ...]:
+        raise NotImplementedError
+
+    def respond(self, iterable: Iterable[T], filename: str) -> StreamingHttpResponse:
+        def row_iter() -> Generator[tuple[str, ...], None, None]:
+            header = self.get_header()
+            if header:
+                yield header
+            for item in iterable:
+                yield self.get_row(item)
+
+        pseudo_buffer = Echo()
+        writer = csv.writer(pseudo_buffer)
+        return StreamingHttpResponse(
+            (writer.writerow(r) for r in row_iter()),
+            content_type="text/csv",
+            headers={"Content-Disposition": f'attachment; filename="{filename}.csv"'},
+        )

src/sentry/web/frontend/group_tag_export.py

@@ -1,29 +1,36 @@
+from __future__ import annotations
+
 from django.http import Http404
+from django.http.response import HttpResponseBase
 from rest_framework.request import Request
-from rest_framework.response import Response
 
 from sentry.api.base import EnvironmentMixin
 from sentry.data_export.base import ExportError
 from sentry.data_export.processors.issues_by_tag import IssuesByTagProcessor
 from sentry.models.environment import Environment
+from sentry.tagstore.types import GroupTagValue
 from sentry.web.frontend.base import ProjectView, region_silo_view
-from sentry.web.frontend.mixins.csv import CsvMixin
+from sentry.web.frontend.csv import CsvResponder
 
 
-@region_silo_view
-class GroupTagExportView(ProjectView, CsvMixin, EnvironmentMixin):
-    required_scope = "event:read"
+class GroupTagCsvResponder(CsvResponder[GroupTagValue]):
+    def __init__(self, key: str) -> None:
+        self.key = key
 
-    def get_header(self, key):
-        return tuple(IssuesByTagProcessor.get_header_fields(key))
+    def get_header(self) -> tuple[str, ...]:
+        return tuple(IssuesByTagProcessor.get_header_fields(self.key))
 
-    def get_row(self, item, key):
-        fields = IssuesByTagProcessor.get_header_fields(key)
-        item_dict = IssuesByTagProcessor.serialize_row(item, key)
-        return (item_dict[field] for field in fields)
+    def get_row(self, item: GroupTagValue) -> tuple[str, ...]:
+        fields = IssuesByTagProcessor.get_header_fields(self.key)
+        item_dict = IssuesByTagProcessor.serialize_row(item, self.key)
+        return tuple(item_dict[field] for field in fields)
 
-    def get(self, request: Request, organization, project, group_id, key) -> Response:
 
+@region_silo_view
+class GroupTagExportView(ProjectView, EnvironmentMixin):
+    required_scope = "event:read"
+
+    def get(self, request: Request, organization, project, group_id, key) -> HttpResponseBase:
         # If the environment doesn't exist then the tag can't possibly exist
         try:
             environment_id = self._get_environment_id_from_request(request, project.organization_id)
@@ -43,4 +50,4 @@ class GroupTagExportView(ProjectView, CsvMixin, EnvironmentMixin):
 
         filename = f"{processor.group.qualified_short_id or processor.group.id}-{key}"
 
-        return self.to_csv_response(processor.get_raw_data(), filename, key=key)
+        return GroupTagCsvResponder(key).respond(processor.get_raw_data(), filename)

src/sentry/web/frontend/mixins/csv.py

@@ -11,6 +11,8 @@ class Echo:
 
 
 class CsvMixin:
+    """deprecated: will be removed! use sentry.web.csv.CsvResponder instead!"""
+
     def get_header(self, **kwargs):
         return ()