use template to send mail for export job completion

docker-compose.yaml

@@ -17,7 +17,7 @@ services:
    - ./pyenv:/home/test
  celery:
   build: .
-  entrypoint: poetry run celery -A project.celery worker
+  entrypoint: poetry run celery -A project.celery worker -l info
   links:
    - postgres
    - redis

example/project/__init__.py

@@ -0,0 +1,3 @@
+from .celery import app as celery_app
+
+__all__ = ("celery_app",)

example/project/settings.py

@@ -58,7 +58,7 @@
 TEMPLATES = [
     {
         "BACKEND": "django.template.backends.django.DjangoTemplates",
-        "DIRS": [],
+        "DIRS": ["templates"],
         "APP_DIRS": True,
         "OPTIONS": {
             "context_processors": [

example/pyproject.toml

@@ -13,6 +13,7 @@ celery = "*"
 django-author = "*"
 redis = "*"
 psycopg2-binary = "^2.8.4"
+html2text = "^2020.1.16"
 
 [tool.poetry.dev-dependencies]
 django-admin-smoke-tests = "^0.3.0"

example/winners/tests/test_utils.py

@@ -0,0 +1,42 @@
+from django.test import TestCase, override_settings
+
+from import_export_celery.utils import (
+    get_export_job_mail_subject,
+    get_export_job_mail_template,
+    get_export_job_mail_context,
+    DEFAULT_EXPORT_JOB_COMPLETION_MAIL_SUBJECT,
+    DEFAULT_EXPORT_JOB_COMPLETION_MAIL_TEMPLATE,
+)
+from import_export_celery.models import ExportJob
+
+
+class UtilsTestCases(TestCase):
+    def test_get_export_job_mail_subject_by_default(self):
+        self.assertEqual(
+            DEFAULT_EXPORT_JOB_COMPLETION_MAIL_SUBJECT, get_export_job_mail_subject()
+        )
+
+    @override_settings(EXPORT_JOB_COMPLETION_MAIL_SUBJECT="New subject")
+    def test_get_export_job_mail_subject_overridden(self):
+        self.assertEqual("New subject", get_export_job_mail_subject())
+
+    def test_get_export_job_mail_template_default(self):
+        self.assertEqual(
+            DEFAULT_EXPORT_JOB_COMPLETION_MAIL_TEMPLATE, get_export_job_mail_template()
+        )
+
+    @override_settings(EXPORT_JOB_COMPLETION_MAIL_TEMPLATE="mytemplate.html")
+    def test_get_export_job_mail_template_overridden(self):
+        self.assertEqual("mytemplate.html", get_export_job_mail_template())
+
+    def test_get_export_job_mail_context(self):
+        export_job = ExportJob.objects.create(
+            app_label="winners", model="Winner", site_of_origin="http://127.0.0.1:8000"
+        )
+        context = get_export_job_mail_context(export_job)
+        expected_context = {
+            "app_label": "winners",
+            "model": "Winner",
+            "link": f"http://127.0.0.1:8000/adminimport_export_celery/exportjob/{export_job.id}/change/",
+        }
+        self.assertEqual(context, expected_context)

import_export_celery/tasks.py

@@ -7,16 +7,15 @@
 from django.conf import settings
 from django.core.files.base import ContentFile
 from django.core.cache import cache
-from django.core.mail import send_mail
 
-from django.urls import reverse
 from django.utils.encoding import force_str
 from django.utils.translation import gettext_lazy as _
 
 from import_export.formats.base_formats import DEFAULT_FORMATS
 
 from . import models
 from .model_config import ModelConfig
+from .utils import send_export_job_completion_mail
 
 from celery.utils.log import get_task_logger
 import logging
@@ -234,24 +233,5 @@ def export_resource(self, *args, **kwargs):
         serialized = serialized.encode("utf8")
     export_job.file.save(filename, ContentFile(serialized))
     if export_job.email_on_completion:
-        send_mail(
-            _("Django: Export job completed"),
-            _(
-                "Your export job on model {app_label}.{model} has completed. You can download the file at the following link:\n\n{link}"
-            ).format(
-                app_label=export_job.app_label,
-                model=export_job.model,
-                link=export_job.site_of_origin
-                + reverse(
-                    "admin:%s_%s_change"
-                    % (
-                        export_job._meta.app_label,
-                        export_job._meta.model_name,
-                    ),
-                    args=[export_job.pk],
-                ),
-            ),
-            settings.SERVER_EMAIL,
-            [export_job.updated_by.email],
-        )
+        send_export_job_completion_mail(export_job)
     return

import_export_celery/templates/email/export_job_completion.html

@@ -0,0 +1,14 @@
+<!DOCTYPE html>
+<html lang="en">
+    <head>
+        <meta charset="UTF-8" />
+        <meta http-equiv="X-UA-Compatible" content="IE=edge" />
+        <meta name="viewport" content="width=device-width, initial-scale=1.0" />
+        <title>Document</title>
+    </head>
+
+    <body>
+        <p>Your export job on model {{app_label}}.{{model}} has completed. You can download the file at the following link:</p>
+        <a href="{{link}}" target="_blank">{{link}}</a>
+    </body>
+</html>

import_export_celery/utils.py

@@ -0,0 +1,70 @@
+import html2text
+from django.core.mail import send_mail
+from django.template.loader import get_template
+from django.conf import settings
+from django.urls import reverse
+
+DEFAULT_EXPORT_JOB_COMPLETION_MAIL_SUBJECT = "Django: Export job completed"
+DEFAULT_EXPORT_JOB_COMPLETION_MAIL_TEMPLATE = "email/export_job_completion.html"
+
+
+def build_html_and_text_message(template_name, context={}):
+    """
+    Render the given template with the context and returns
+    the data in html and plain text format.
+    """
+    template = get_template(template_name)
+    html_message = template.render(context)
+    text_message = html2text.html2text(html_message)
+    return html_message, text_message
+
+
+def get_export_job_mail_context(export_job):
+    context = {
+        "app_label": export_job.app_label,
+        "model": export_job.model,
+        "link": export_job.site_of_origin
+        + reverse(
+            "admin:%s_%s_change"
+            % (
+                export_job._meta.app_label,
+                export_job._meta.model_name,
+            ),
+            args=[export_job.pk],
+        ),
+    }
+    return context
+
+
+def get_export_job_mail_subject():
+    return getattr(
+        settings,
+        "EXPORT_JOB_COMPLETION_MAIL_SUBJECT",
+        DEFAULT_EXPORT_JOB_COMPLETION_MAIL_SUBJECT,
+    )
+
+
+def get_export_job_mail_template():
+    return getattr(
+        settings,
+        "EXPORT_JOB_COMPLETION_MAIL_TEMPLATE",
+        DEFAULT_EXPORT_JOB_COMPLETION_MAIL_TEMPLATE,
+    )
+
+
+def send_export_job_completion_mail(export_job):
+    """
+    Send export job completion mail
+    """
+    subject = get_export_job_mail_subject()
+    template_name = get_export_job_mail_template()
+    context = get_export_job_mail_context(export_job)
+    context.update({"export_job": export_job})
+    html_message, text_message = build_html_and_text_message(template_name, context)
+    send_mail(
+        subject=subject,
+        message=text_message,
+        html_message=html_message,
+        from_email=settings.SERVER_EMAIL,
+        recipient_list=[export_job.updated_by.email],
+    )

setup.cfg

@@ -1,6 +1,7 @@
 [flake8]
 max-line-length = 150
 exclude =
+   pyenv/*,
    env/*,
    env-dj19/*,
    *migrations/*,

setup.py

@@ -6,7 +6,7 @@
 
 here = os.path.abspath(os.path.dirname(__file__))
 
-requires = ["Django", "django-import-export", "django-author"]
+requires = ["Django", "django-import-export", "django-author", "html2text"]
 
 try:
     version = subprocess.check_output(