feat: generate coupon pdfs in the background

poukazky/app/admin.py

@@ -3,16 +3,19 @@
 from django import forms
 from django.contrib import admin, messages
 from django.db import IntegrityError, transaction
+from django.db.models import QuerySet
 from django.forms import ValidationError
-from django.http import HttpResponse
+from django.http import HttpResponseRedirect
+from django.urls import reverse
 from django_admin_action_forms.admin import AdminActionFormsMixin
 from django_admin_action_forms.decorators import action_with_form
 from django_admin_action_forms.forms import AdminActionForm
 from django_no_queryset_admin_actions.admin import NoQuerySetAdminActionsMixin
 from django_no_queryset_admin_actions.decorators import no_queryset_action
+from rq.job import Job
 
 from poukazky.app.models import ExternalCoupon, Provider, TrojstenCoupon
-from poukazky.app.utils import generate_coupons
+from poukazky.app.tasks import generate_coupons_pdf
 
 
 class GenerateCouponForm(AdminActionForm):
@@ -59,30 +62,22 @@ def is_used(self, value):
     def generate_coupon(self, request, data):
         with transaction.atomic():
             coupons = [
-                TrojstenCoupon.generate(data["amount"], data["expires_at"])
+                TrojstenCoupon.generate(data["amount"], data["expires_at"]).id
                 for _ in range(data["count"])
             ]
 
-        output = generate_coupons(coupons)
-
-        return HttpResponse(
-            output,
-            content_type="application/pdf",
-            headers={
-                "Content-Disposition": 'attachment; filename="trojsten-poukazky.pdf"'
-            },
+        job: Job = generate_coupons_pdf.delay(coupons)  # type:ignore
+        return HttpResponseRedirect(
+            reverse("coupon_pdf_download", kwargs={"job_id": job.id})
         )
 
     @admin.display(description="Vygenerovať PDF označených poukážok")
-    def regenerate_coupons(self, request, queryset):
-        output = generate_coupons(queryset.all())
-
-        return HttpResponse(
-            output,
-            content_type="application/pdf",
-            headers={
-                "Content-Disposition": 'attachment; filename="trojsten-poukazky.pdf"'
-            },
+    def regenerate_coupons(self, request, queryset: QuerySet[TrojstenCoupon]):
+        job: Job = generate_coupons_pdf.delay(
+            list(queryset.values_list("id", flat=True))
+        )  # type:ignore
+        return HttpResponseRedirect(
+            reverse("coupon_pdf_download", kwargs={"job_id": job.id})
         )
 
     @action_with_form(

poukazky/app/tasks.py

@@ -1,7 +1,13 @@
+import secrets
+from datetime import timedelta
+from io import BytesIO
+
+from django.core.files.storage import default_storage
+from django.utils import timezone
 from django_rq import job
 
-from poukazky.app.models import ExternalCoupon
-from poukazky.app.utils import send_mail
+from poukazky.app.models import ExternalCoupon, TrojstenCoupon
+from poukazky.app.utils import generate_coupons, send_mail
 
 
 @job
@@ -13,3 +19,25 @@ def coupon_exchanged(coupon_id: int):
         "coupon_exchanged",
         context={"coupon": coupon},
     )
+
+
+@job
+def generate_coupons_pdf(coupon_ids: list[int]) -> str:
+    coupons = TrojstenCoupon.objects.filter(id__in=coupon_ids)
+    pdf_bytes = generate_coupons(coupons)
+
+    filename = f"tmp_coupons/{secrets.token_hex(32)}.pdf"
+    default_storage.save(filename, BytesIO(pdf_bytes))
+    cleanup_coupon_pdfs.delay()
+    return filename
+
+
+@job
+def cleanup_coupon_pdfs():
+    cutoff = timezone.now() - timedelta(minutes=30)
+    _, filenames = default_storage.listdir("tmp_coupons")
+
+    for filename in filenames:
+        path = f"tmp_coupons/{filename}"
+        if default_storage.get_modified_time(path) < cutoff:
+            default_storage.delete(path)

poukazky/app/urls.py

@@ -1,6 +1,11 @@
 from django.urls import path
 
-from poukazky.app.views import CouponDetailView, CouponExchangeView, CouponFormView
+from poukazky.app.views import (
+    CouponDetailView,
+    CouponExchangeView,
+    CouponFormView,
+    CouponPDFView,
+)
 
 urlpatterns = [
     path("", CouponFormView.as_view(), name="coupon_form"),
@@ -10,4 +15,9 @@
         CouponExchangeView.as_view(),
         name="coupon_exchange",
     ),
+    path(
+        "download_pdf/<job_id>/",
+        CouponPDFView.as_view(),
+        name="coupon_pdf_download",
+    ),
 ]

poukazky/app/utils.py

@@ -1,3 +1,5 @@
+from typing import Iterable
+
 from django.conf import settings
 from django.core.mail import EmailMultiAlternatives
 from django.template.loader import render_to_string
@@ -6,7 +8,7 @@
 from poukazky.app.render_typst import render_typst
 
 
-def generate_coupons(coupons: list[TrojstenCoupon]) -> bytes:
+def generate_coupons(coupons: Iterable[TrojstenCoupon]) -> bytes:
     ctx = {}
 
     ctx["coupons"] = [

poukazky/app/views.py

@@ -1,16 +1,21 @@
 from typing import Any
 
+import django_rq
 from django.conf import settings
+from django.contrib.auth.mixins import PermissionRequiredMixin
+from django.core.files.storage import default_storage
 from django.db import transaction
 from django.db.models import Exists, OuterRef
-from django.http import HttpResponseRedirect
+from django.http import Http404, HttpResponse, HttpResponseRedirect
 from django.shortcuts import get_object_or_404, redirect
 from django.urls import reverse
 from django.utils import timezone
 from django.utils.functional import cached_property
 from django.utils.timezone import timedelta
 from django.views import View
 from django.views.generic import DetailView, FormView
+from rq.exceptions import NoSuchJobError
+from rq.job import Job, JobStatus
 
 from poukazky.app.forms import CouponExchangeForm, CouponSearchForm
 from poukazky.app.models import ExternalCoupon, Provider, TrojstenCoupon
@@ -136,3 +141,23 @@ def form_valid(self, form):
             coupon_exchanged.delay(external_coupon.id)
 
         return redirect("coupon_detail", code=self.coupon.code)
+
+
+class CouponPDFView(PermissionRequiredMixin, View):
+    permission_required = "app.create_trojstencoupon"
+
+    def get(self, request, *args, **kwargs):
+        conn = django_rq.get_connection()
+        try:
+            job = Job.fetch(kwargs["job_id"], conn)
+            if job.func_name != "poukazky.app.tasks.generate_coupons_pdf":
+                raise Http404()
+        except NoSuchJobError:
+            raise Http404()
+
+        if job.get_status() != JobStatus.FINISHED:
+            resp = HttpResponse("Rendering...")
+            resp.headers["Refresh"] = 5
+            return resp
+
+        return HttpResponseRedirect(default_storage.url(job.result))