Django Admin action for exporting selected rows as CSV#

I wanted to add an action option to the Django Admin for exporting the currently selected set of rows (or every row in the table) as a CSV file.

I ended up using a pattern inspired by this Django Snippet, but with an added touch for more efficient exports. In order to avoid using up too much memory for the export, I use keyset pagination to fetch 500 rows at a time.

The keyset_pagination_iterator() helper function accepts any queryset, orders it by the primary key and then repeatedly fetches 500 items. It then modifies the queryset to add a WHERE id > $last_seen_id clause. This is a relatively inexpensive way to paginate, so having an endpoint perform that query dozens or even hundreds of times should hopefully avoid adding too much load to the database.

The action itself uses a pattern that combines StringIO and csv.writer() to stream out the results as a CSV file.

Django’s StreamingHttpResponse mechanism is really neat: it accepts a Python iterator or generator and returns a streaming response derived from that sequence.

The Django documentation says “Streaming responses will tie a worker process for the entire duration of the response. This may result in poor performance” - this particular project runs on Google Cloud Run so I’m less concerned about tying up a worker than I would be normally, plus the export option is only available to trusted staff users with access to the Django Admin interface.

To add the CSV export option to a ModelAdmin subclass, do the following:

1
from .admin_actions import export_as_csv_action
2

3
@admin.register(County)
4
class CountyAdmin(admin.ModelAdmin):
5
    actions = [export_as_csv_action()]

Here’s admin_actions.py:

1
import csv
2
from io import StringIO
3

4
from django.http import StreamingHttpResponse
5

6

7
def keyset_pagination_iterator(input_queryset, batch_size=500):
8
    all_queryset = input_queryset.order_by("pk")
9
    last_pk = None
10
    while True:
11
        queryset = all_queryset
12
        if last_pk is not None:
13
            queryset = all_queryset.filter(pk__gt=last_pk)
14
        queryset = queryset[:batch_size]
15
        for row in queryset:
16
            last_pk = row.pk
17
            yield row
18
        if not queryset:
19
            break
20

21

22
def export_as_csv_action(description="Export selected rows to CSV"):
23
    def export_as_csv(modeladmin, request, queryset):
24
        def rows(queryset):
25

26
            csvfile = StringIO()
27
            csvwriter = csv.writer(csvfile)
28
            columns = [field.name for field in modeladmin.model._meta.fields]
29

30
            def read_and_flush():
31
                csvfile.seek(0)
32
                data = csvfile.read()
33
                csvfile.seek(0)
34
                csvfile.truncate()
35
                return data
36

37
            header = False
38

39
            if not header:
40
                header = True
41
                csvwriter.writerow(columns)
42
                yield read_and_flush()
43

44
            for row in keyset_pagination_iterator(queryset):
45
                csvwriter.writerow(getattr(row, column) for column in columns)
46
                yield read_and_flush()
47

48
        response = StreamingHttpResponse(rows(queryset), content_type="text/csv")
49
        response["Content-Disposition"] = (
50
            "attachment; filename=%s.csv" % modeladmin.model.__name__
51
        )
52

53
        return response
54

55
    export_as_csv.short_description = description
56
    return export_as_csv

Newsletter

Django Admin action for exporting selected rows as CSV#