Responses, Streaming, CSV & PDF

from http import HTTPStatus

from django.http import HttpResponse
from django.utils.cache import patch_cache_control


class HttpResponseNoContent(HttpResponse):
    status_code = HTTPStatus.NO_CONTENT


def health(request):
    response = HttpResponse(
        "ok",
        content_type="text/plain",
        headers={"X-Bookshelf-Version": "1"},
    )
    response.headers["X-Checked"] = "yes"          # dict-like access (3.2+)
    patch_cache_control(response, max_age=30, public=True)
    return response


def ping(request):
    return HttpResponseNoContent()


def accepted(request):
    return HttpResponse("queued", status=HTTPStatus.ACCEPTED)

$ curl -si http://127.0.0.1:8101/catalog/health/
HTTP/1.1 200 OK
X-Bookshelf-Version: 1
Content-Type: text/plain
X-Checked: yes
Cache-Control: max-age=30, public

ok

from django.core.serializers.json import DjangoJSONEncoder
from django.http import Http404, JsonResponse
from django.views.decorators.http import require_GET

from .models import Book


def book_to_dict(book):
    return {
        "uid": book.uid,                  # UUID
        "title": book.title,
        "price": book.price,              # Decimal
        "published": book.published,      # date
        "author": book.author.name,
    }


@require_GET
def book_list(request):
    books = Book.objects.select_related("author")
    return JsonResponse({"results": [book_to_dict(b) for b in books]})


@require_GET
def book_detail(request, pk):
    try:
        book = Book.objects.select_related("author").get(pk=pk)
    except Book.DoesNotExist:
        return JsonResponse({"detail": "Not found."}, status=404)
    return JsonResponse(book_to_dict(book), json_dumps_params={"indent": 2})


@require_GET
def isbn_list(request):
    # a top-level array needs safe=False (a dict is preferred; see the text)
    isbns = list(Book.objects.values_list("isbn", flat=True))
    return JsonResponse(isbns, safe=False)


class StrictEncoder(DjangoJSONEncoder):
    """Serialize sets too, as sorted lists."""

    def default(self, o):
        if isinstance(o, (set, frozenset)):
            return sorted(o)
        return super().default(o)


@require_GET
def countries(request):
    return JsonResponse({"countries": {"ES", "FR", "DE"}}, encoder=StrictEncoder)

$ curl -si http://127.0.0.1:8101/catalog/api/books/1/
HTTP/1.1 200 OK
Content-Type: application/json
Content-Length: 161

{
  "uid": "6275540d-c5df-42af-affe-e498c244e233",
  "title": "Don Quijote",
  "price": "12.50",
  "published": "1605-01-16",
  "author": "Miguel de Cervantes"
}

import io

from django.contrib.auth.decorators import login_required
from django.http import FileResponse, Http404, HttpResponse
from django.shortcuts import get_object_or_404

from .models import Book


def sample_download(request, pk):
    """Serve a book's sample chapter stored in its FileField."""
    book = get_object_or_404(Book, pk=pk)
    if not book.sample:
        raise Http404("This book has no sample.")
    # FieldFile.open() returns a file object; FileResponse closes it for us.
    return FileResponse(
        book.sample.open("rb"),
        as_attachment=True,
        filename=f"{book.isbn}-sample.pdf",
    )


def sample_inline(request, pk):
    book = get_object_or_404(Book, pk=pk)
    return FileResponse(book.sample.open("rb"), filename="sample.pdf")


def readme(request):
    buffer = io.BytesIO("Bookshelf export\nLibro: El Quijote\n".encode("utf-8"))
    # buffer is already at position 0; seek(0) is mandatory after writing.
    return FileResponse(buffer, as_attachment=True, filename="Ünïcode readme.txt")


@login_required
def private_sample(request, pk):
    """Let the web server send the file once Django has authorized it."""
    book = get_object_or_404(Book, pk=pk)
    response = HttpResponse()
    response["Content-Disposition"] = f'attachment; filename="{book.isbn}-sample.pdf"'
    response["X-Accel-Redirect"] = f"/protected-samples/{book.sample.name}"
    return response

$ curl -si http://127.0.0.1:8101/catalog/books/1/sample/
HTTP/1.1 200 OK
Content-Type: application/pdf
Content-Length: 15
Content-Disposition: attachment; filename="9780142437230-sample.pdf"

$ curl -sI http://127.0.0.1:8102/catalog/readme.txt
HTTP/1.1 200 OK
Content-Type: text/plain
Content-Length: 35
Content-Disposition: attachment; filename*=utf-8''%C3%9Cn%C3%AFcode%20readme.txt

import csv
from itertools import batched

from django.http import HttpResponse, StreamingHttpResponse
from django.template import loader

from .models import Book

HEADER = ["isbn", "title", "author", "price", "published"]


def safe_cell(value):
    """Neutralize spreadsheet formula injection (CSV injection)."""
    text = "" if value is None else str(value)
    if text[:1] in ("=", "+", "-", "@", "\t", "\r"):
        return "'" + text
    return text


def book_row(book):
    return [
        book.isbn,
        safe_cell(book.title),
        safe_cell(book.author.name),
        book.price,
        book.published or "",
    ]


def books_csv(request):
    response = HttpResponse(
        content_type="text/csv; charset=utf-8",
        headers={"Content-Disposition": 'attachment; filename="books.csv"'},
    )
    response.write("")   # BOM so Excel detects UTF-8
    writer = csv.writer(response)
    writer.writerow(HEADER)
    for book in Book.objects.select_related("author"):
        writer.writerow(book_row(book))
    return response


def books_csv_template(request):
    response = HttpResponse(
        loader.render_to_string(
            "catalog/books.csv", {"books": Book.objects.select_related("author")}
        ),
        content_type="text/csv",
        headers={"Content-Disposition": 'attachment; filename="books-template.csv"'},
    )
    return response


class Echo:
    """An object that implements just the write method of the file-like interface."""

    def write(self, value):
        return value


def books_csv_stream(request):
    writer = csv.writer(Echo())
    queryset = Book.objects.select_related("author").iterator(chunk_size=500)

    def rows():
        yield writer.writerow(HEADER)
        for batch in batched(queryset, 100):
            yield "".join(writer.writerow(book_row(book)) for book in batch)

    return StreamingHttpResponse(
        rows(),
        content_type="text/csv",
        headers={"Content-Disposition": 'attachment; filename="books-stream.csv"'},
    )

isbn,title,author,price,published
9780000000002,"'=HYPERLINK(""http://evil"")",Miguel de Cervantes,1.00,
9780142437230,Don Quijote,Miguel de Cervantes,12.50,1605-01-16
9780000000003,"The ""Big"", Book",Miguel de Cervantes,2.00,

import asyncio
import io

from django.http import FileResponse
from reportlab.lib import colors
from reportlab.lib.pagesizes import A4
from reportlab.lib.styles import getSampleStyleSheet
from reportlab.platypus import Paragraph, SimpleDocTemplate, Spacer, Table, TableStyle

from .models import Book


def build_catalog_pdf(rows):
    """Pure function: list of (isbn, title, author, price) -> PDF bytes buffer."""
    buffer = io.BytesIO()
    doc = SimpleDocTemplate(buffer, pagesize=A4, title="Bookshelf catalogue")
    styles = getSampleStyleSheet()
    table = Table([["ISBN", "Title", "Author", "Price"], *rows], repeatRows=1)
    table.setStyle(
        TableStyle(
            [
                ("BACKGROUND", (0, 0), (-1, 0), colors.lightgrey),
                ("GRID", (0, 0), (-1, -1), 0.25, colors.grey),
                ("ALIGN", (3, 1), (3, -1), "RIGHT"),
            ]
        )
    )
    doc.build([Paragraph("Bookshelf catalogue", styles["Title"]), Spacer(1, 12), table])
    buffer.seek(0)
    return buffer


def catalog_pdf(request):
    rows = [
        (b.isbn, b.title, b.author.name, f"{b.price:.2f}")
        for b in Book.objects.select_related("author")
    ]
    return FileResponse(
        build_catalog_pdf(rows), as_attachment=True, filename="catalogue.pdf"
    )


async def catalog_pdf_async(request):
    rows = [
        (b.isbn, b.title, b.author.name, f"{b.price:.2f}")
        async for b in Book.objects.select_related("author")
    ]
    # ReportLab is blocking and CPU-bound: keep it off the event loop.
    buffer = await asyncio.to_thread(build_catalog_pdf, rows)
    return FileResponse(buffer, as_attachment=True, filename="catalogue.pdf")

from django.template.response import TemplateResponse

from .models import Book


def with_compact_option(view):
    """Decorator: swap the template after the view ran, thanks to TemplateResponse."""

    def wrapper(request, *args, **kwargs):
        response = view(request, *args, **kwargs)
        if request.GET.get("compact") and hasattr(response, "template_name"):
            response.template_name = "catalog/book_list_compact.html"
        return response

    return wrapper


@with_compact_option
def book_page(request):
    response = TemplateResponse(
        request,
        "catalog/book_list.html",
        {"books": Book.objects.select_related("author")},
        headers={"X-Page": "book-list"},
    )
    response.add_post_render_callback(lambda r: r.headers.setdefault("X-Rendered", "1"))
    return response

class SiteContextMiddleware:
    """Add shared context to every TemplateResponse, after the view returned."""

    def __init__(self, get_response):
        self.get_response = get_response

    def __call__(self, request):
        return self.get_response(request)

    def process_template_response(self, request, response):
        if response.context_data is not None:
            response.context_data["site_name"] = "Bookshelf"
        return response

<h1>{{ site_name|default:"?" }}: {{ books|length }} books</h1>
<ul>
{% for book in books %}  <li>{{ book.title }} by {{ book.author }}</li>
{% endfor %}</ul>

{{ site_name }}: {% for book in books %}{{ book.title }}{% if not forloop.last %}, {% endif %}{% endfor %}

$ curl -si http://127.0.0.1:8101/catalog/books/
HTTP/1.1 200 OK
X-Page: book-list
Content-Type: text/html; charset=utf-8
X-Rendered: 1

<h1>Bookshelf: 3 books</h1>
<ul>
  <li>Don Quijote by Miguel de Cervantes</li>
  <li>El Quijote II by Miguel de Cervantes</li>
  <li>La Galatea by Miguel de Cervantes</li>
</ul>

$ curl -s "http://127.0.0.1:8101/catalog/books/?compact=1"
Bookshelf: Don Quijote, El Quijote II, La Galatea

import csv
import io
from decimal import Decimal

from django.http import JsonResponse
from django.test import AsyncClient, TestCase

from .models import Author, Book


class ResponseTests(TestCase):
    @classmethod
    def setUpTestData(cls):
        author = Author.objects.create(name="Miguel de Cervantes")
        cls.book = Book.objects.create(
            title="Don Quijote", isbn="9780142437230", author=author, price=Decimal("12.50")
        )
        Book.objects.create(
            title='=HYPERLINK("http://evil")', isbn="9780000000002", author=author, price=1
        )

    def test_json_response(self):
        response = self.client.get(f"/catalog/api/books/{self.book.pk}/")
        self.assertIsInstance(response, JsonResponse)
        self.assertEqual(response["Content-Type"], "application/json")
        self.assertEqual(response.json()["price"], "12.50")   # Decimal -> string

    def test_json_rejects_non_dict_by_default(self):
        with self.assertRaises(TypeError):
            JsonResponse([1, 2, 3])

    def test_csv_download(self):
        response = self.client.get("/catalog/export/books.csv")
        self.assertEqual(response["Content-Type"], "text/csv; charset=utf-8")
        self.assertIn("attachment", response["Content-Disposition"])
        rows = list(csv.reader(io.StringIO(response.content.decode("utf-8-sig"))))
        self.assertEqual(rows[0][:2], ["isbn", "title"])
        self.assertIn("'=HYPERLINK(\"http://evil\")", [row[1] for row in rows])

    def test_streaming_csv(self):
        response = self.client.get("/catalog/export/books-stream.csv")
        self.assertTrue(response.streaming)
        body = b"".join(response.streaming_content).decode()
        self.assertEqual(body.count("\r\n"), 3)               # header + 2 rows

    def test_pdf(self):
        response = self.client.get("/catalog/export/catalogue.pdf")
        self.assertEqual(response["Content-Type"], "application/pdf")
        self.assertTrue(b"".join(response.streaming_content).startswith(b"%PDF-"))

    def test_template_response_is_inspectable(self):
        response = self.client.get("/catalog/books/?compact=1")
        self.assertEqual(response.template_name, "catalog/book_list_compact.html")
        self.assertEqual(response.context_data["site_name"], "Bookshelf")
        self.assertContains(response, "Don Quijote")


class AsyncStreamingTests(TestCase):
    async def test_async_stream_and_sse_headers(self):
        client = AsyncClient()
        response = await client.get("/catalog/countdown/async/")
        self.assertTrue(response.is_async)
        chunks = [chunk async for chunk in response.streaming_content]
        self.assertEqual(b"".join(chunks), b"3\n2\n1\nliftoff\n")

        events = await client.get("/catalog/events/books/")
        self.assertEqual(events["Content-Type"], "text/event-stream")
        self.assertEqual(events["Cache-Control"], "no-cache")
        stream = events.streaming_content.__aiter__()
        self.assertEqual(await stream.__anext__(), b"retry: 3000\n\n")
        await stream.aclose()