Refactor email generation to use a class-based design like Django views
This commit is contained in:
parent
a7e7fafedd
commit
7ce28d449f
81
cmsmanage/email.py
Normal file
81
cmsmanage/email.py
Normal file
@ -0,0 +1,81 @@
|
||||
from collections.abc import Sequence
|
||||
from typing import Any, ClassVar, TypeAlias
|
||||
|
||||
from django.core import mail
|
||||
from django.template import loader
|
||||
|
||||
import mdformat
|
||||
from markdownify import markdownify
|
||||
|
||||
Context: TypeAlias = dict[str, Any]
|
||||
|
||||
|
||||
class EmailBase:
|
||||
subject: ClassVar[str]
|
||||
from_email: ClassVar[str | None] = None
|
||||
reply_to: ClassVar[Sequence[str] | None] = None
|
||||
|
||||
context: Context
|
||||
|
||||
def __init__(self, context: Context) -> None:
|
||||
self.context = context
|
||||
|
||||
def render_body(self) -> str:
|
||||
raise NotImplementedError
|
||||
|
||||
@classmethod
|
||||
def render(
|
||||
cls,
|
||||
context: Context,
|
||||
to: list[str] | None = None,
|
||||
cc: list[str] | None = None,
|
||||
bcc: list[str] | None = None,
|
||||
) -> mail.EmailMessage:
|
||||
self = cls(context)
|
||||
|
||||
body = self.render_body()
|
||||
|
||||
return mail.EmailMessage(
|
||||
self.subject,
|
||||
body,
|
||||
from_email=self.from_email,
|
||||
reply_to=self.reply_to,
|
||||
to=to,
|
||||
cc=cc,
|
||||
bcc=bcc,
|
||||
)
|
||||
|
||||
|
||||
class TemplatedMultipartEmail(EmailBase):
|
||||
template: ClassVar[str]
|
||||
|
||||
def render_html_body(self) -> str:
|
||||
template = loader.get_template(self.template)
|
||||
return template.render(self.context)
|
||||
|
||||
def render_body(self, html_body: str) -> str:
|
||||
return mdformat.text(markdownify(html_body), extensions={"tables"})
|
||||
|
||||
@classmethod
|
||||
def render(
|
||||
cls,
|
||||
context: Context,
|
||||
to: list[str] | None = None,
|
||||
cc: list[str] | None = None,
|
||||
bcc: list[str] | None = None,
|
||||
) -> mail.EmailMessage:
|
||||
self = cls(context)
|
||||
|
||||
html_body = self.render_html_body()
|
||||
plain_body = self.render_body(html_body)
|
||||
|
||||
return mail.EmailMultiAlternatives(
|
||||
self.subject,
|
||||
plain_body,
|
||||
from_email=self.from_email,
|
||||
reply_to=self.reply_to,
|
||||
to=to,
|
||||
cc=cc,
|
||||
bcc=bcc,
|
||||
alternatives=[(html_body, "text/html")],
|
||||
)
|
||||
@ -1,67 +1,62 @@
|
||||
from django.conf import settings
|
||||
from django.core.mail import EmailMessage, EmailMultiAlternatives
|
||||
from django.template import loader
|
||||
|
||||
import mdformat
|
||||
from markdownify import markdownify
|
||||
from django.core.mail import EmailMessage
|
||||
|
||||
from cmsmanage.email import TemplatedMultipartEmail
|
||||
from membershipworks.models import EventInvoice
|
||||
|
||||
|
||||
def make_multipart_email(
|
||||
subject: str, html_body: str, to: tuple[str]
|
||||
) -> EmailMultiAlternatives:
|
||||
plain_body = mdformat.text(markdownify(html_body), extensions={"tables"})
|
||||
|
||||
email = EmailMultiAlternatives(
|
||||
subject,
|
||||
plain_body,
|
||||
from_email="CMS Invoices <invoices@claremontmakerspace.org>",
|
||||
to=to,
|
||||
reply_to=["Claremont MakerSpace <Info@ClaremontMakerSpace.org>"],
|
||||
)
|
||||
|
||||
email.attach_alternative(html_body, "text/html")
|
||||
return email
|
||||
class InvoiceEmailBase(TemplatedMultipartEmail):
|
||||
from_email = "CMS Invoices <invoices@claremontmakerspace.org>"
|
||||
reply_to = ["Claremont MakerSpace <Info@ClaremontMakerSpace.org>"]
|
||||
|
||||
|
||||
def make_instructor_email(
|
||||
invoice: EventInvoice, pdf: bytes, event_url: str
|
||||
) -> EmailMessage:
|
||||
if invoice.event.instructor is None or invoice.event.instructor.member is None:
|
||||
raise ValueError("Event Instructor not defined or is not member")
|
||||
template = loader.get_template(
|
||||
"membershipworks/email/event_invoice_instructor.dj.html"
|
||||
)
|
||||
html_body = template.render({"invoice": invoice, "event_url": event_url})
|
||||
class InstructorInvoiceEmail(InvoiceEmailBase):
|
||||
template = "membershipworks/email/event_invoice_instructor.dj.html"
|
||||
|
||||
message = make_multipart_email(
|
||||
f'Your CMS instructor invoice has been received for event "{invoice.event}" {invoice.event.start} - {invoice.event.end}',
|
||||
html_body,
|
||||
(invoice.event.instructor.member.sanitized_mailbox(),),
|
||||
)
|
||||
message.attach(f"CMS_event_invoice_{invoice.uuid}.pdf", pdf, "application/pdf")
|
||||
return message
|
||||
@property
|
||||
def subject(self) -> str:
|
||||
event = self.context["invoice"].event
|
||||
return f'Your CMS instructor invoice has been received for event "{event}" {event.start} - {event.end}'
|
||||
|
||||
@classmethod
|
||||
def render_for_invoice(
|
||||
cls, invoice: EventInvoice, pdf: bytes, event_url: str
|
||||
) -> EmailMessage:
|
||||
if invoice.event.instructor is None or invoice.event.instructor.member is None:
|
||||
raise ValueError("Event Instructor not defined or is not member")
|
||||
message = cls.render(
|
||||
{"invoice": invoice, "event_url": event_url},
|
||||
to=[invoice.event.instructor.member.sanitized_mailbox()],
|
||||
)
|
||||
message.attach(f"CMS_event_invoice_{invoice.uuid}.pdf", pdf, "application/pdf")
|
||||
return message
|
||||
|
||||
|
||||
def make_admin_email(invoice: EventInvoice, pdf: bytes, event_url: str) -> EmailMessage:
|
||||
template = loader.get_template("membershipworks/email/event_invoice_admin.dj.html")
|
||||
html_body = template.render({"invoice": invoice, "event_url": event_url})
|
||||
class AdminInvoiceEmail(InvoiceEmailBase):
|
||||
template = "membershipworks/email/event_invoice_admin.dj.html"
|
||||
|
||||
message = make_multipart_email(
|
||||
f'CMS instructor invoice created for event "{invoice.event}" {invoice.event.start} - {invoice.event.end}',
|
||||
html_body,
|
||||
# TODO: should this be in database instead?
|
||||
settings.INVOICE_HANDLERS,
|
||||
)
|
||||
message.attach(f"CMS_event_invoice_{invoice.uuid}.pdf", pdf, "application/pdf")
|
||||
return message
|
||||
@property
|
||||
def subject(self) -> str:
|
||||
event = self.context["invoice"].event
|
||||
return f'CMS instructor invoice created for event "{event}" {event.start} - {event.end}'
|
||||
|
||||
@classmethod
|
||||
def render_for_invoice(
|
||||
cls, invoice: EventInvoice, pdf: bytes, event_url: str
|
||||
) -> EmailMessage:
|
||||
message = cls.render(
|
||||
{"invoice": invoice, "event_url": event_url},
|
||||
# TODO: should this be in database instead?
|
||||
to=settings.INVOICE_HANDLERS,
|
||||
)
|
||||
message.attach(f"CMS_event_invoice_{invoice.uuid}.pdf", pdf, "application/pdf")
|
||||
return message
|
||||
|
||||
|
||||
def make_invoice_emails(
|
||||
invoice: EventInvoice, pdf: bytes, event_url: str
|
||||
) -> list[EmailMessage]:
|
||||
return [
|
||||
make_instructor_email(invoice, pdf, event_url),
|
||||
make_admin_email(invoice, pdf, event_url),
|
||||
InstructorInvoiceEmail.render_for_invoice(invoice, pdf, event_url),
|
||||
AdminInvoiceEmail.render_for_invoice(invoice, pdf, event_url),
|
||||
]
|
||||
|
||||
@ -1,93 +1,114 @@
|
||||
from abc import abstractmethod
|
||||
from collections.abc import Iterable, Iterator
|
||||
from itertools import groupby
|
||||
|
||||
from django.contrib.auth import get_user_model
|
||||
from django.core import mail
|
||||
from django.core.mail.message import sanitize_address
|
||||
from django.template import loader
|
||||
from django.db.models import QuerySet
|
||||
|
||||
import mdformat
|
||||
from markdownify import markdownify
|
||||
from cmsmanage.email import TemplatedMultipartEmail
|
||||
from membershipworks.models import Member
|
||||
from paperwork.models import Certification, Department
|
||||
|
||||
|
||||
def make_multipart_email(subject, html_body, to):
|
||||
plain_body = mdformat.text(markdownify(html_body), extensions={"tables"})
|
||||
class CertificationEmailBase(TemplatedMultipartEmail):
|
||||
from_email = "Claremont MakerSpace Member Certification System <Certifications@ClaremontMakerSpace.org>"
|
||||
reply_to = ["Claremont MakerSpace <Info@ClaremontMakerSpace.org>"]
|
||||
|
||||
email = mail.EmailMultiAlternatives(
|
||||
subject,
|
||||
plain_body,
|
||||
"Claremont MakerSpace Member Certification System <Certifications@ClaremontMakerSpace.org>",
|
||||
to,
|
||||
reply_to=["Claremont MakerSpace <Info@ClaremontMakerSpace.org>"],
|
||||
)
|
||||
|
||||
email.attach_alternative(html_body, "text/html")
|
||||
return email
|
||||
@classmethod
|
||||
@abstractmethod
|
||||
def render_for_certifications(
|
||||
cls, ordered_certifications: QuerySet[Certification]
|
||||
) -> Iterable[mail.EmailMessage]:
|
||||
raise NotImplementedError
|
||||
|
||||
|
||||
def make_department_email(department, certifications):
|
||||
template = loader.get_template("paperwork/email/department_certifications.dj.html")
|
||||
shop_leads = department.shop_lead_flag.members.values_list(
|
||||
"first_name", "account_name", "email", named=True
|
||||
)
|
||||
class DepartmentCertificationEmail(CertificationEmailBase):
|
||||
template = "paperwork/email/department_certifications.dj.html"
|
||||
|
||||
html_body = template.render(
|
||||
{
|
||||
"shop_lead_names": [shop_lead.first_name for shop_lead in shop_leads],
|
||||
"department": department,
|
||||
"certifications": certifications,
|
||||
}
|
||||
)
|
||||
@property
|
||||
def subject(self) -> str:
|
||||
certification_count = len(self.context["certifications"])
|
||||
return f"{certification_count} new CMS Certifications issued for {self.context['department']}"
|
||||
|
||||
return make_multipart_email(
|
||||
f"{len(certifications)} new CMS Certifications issued for {department}",
|
||||
html_body,
|
||||
to=[
|
||||
sanitize_address((shop_lead.account_name, shop_lead.email), "ascii")
|
||||
for shop_lead in shop_leads
|
||||
],
|
||||
)
|
||||
|
||||
|
||||
def department_emails(ordered_queryset):
|
||||
certifications_by_department = groupby(
|
||||
ordered_queryset, lambda c: c.certification_version.definition.department
|
||||
)
|
||||
|
||||
for department, certifications in certifications_by_department:
|
||||
@classmethod
|
||||
def render_for_department(
|
||||
cls, department: Department, certifications: Iterator[Certification]
|
||||
) -> Iterable[mail.EmailMessage]:
|
||||
if department.shop_lead_flag is not None:
|
||||
yield make_department_email(department, list(certifications))
|
||||
shop_leads = department.shop_lead_flag.members.values_list(
|
||||
"first_name", "account_name", "email", named=True
|
||||
)
|
||||
yield cls.render(
|
||||
{
|
||||
"shop_lead_names": [
|
||||
shop_lead.first_name for shop_lead in shop_leads
|
||||
],
|
||||
"department": department,
|
||||
"certifications": list(certifications),
|
||||
},
|
||||
to=[
|
||||
sanitize_address((shop_lead.account_name, shop_lead.email), "ascii")
|
||||
for shop_lead in shop_leads
|
||||
],
|
||||
)
|
||||
|
||||
@classmethod
|
||||
def render_for_certifications(
|
||||
cls, ordered_certifications: QuerySet[Certification]
|
||||
) -> Iterable[mail.EmailMessage]:
|
||||
certifications_by_department = groupby(
|
||||
ordered_certifications,
|
||||
lambda c: c.certification_version.definition.department,
|
||||
)
|
||||
|
||||
for department, certifications in certifications_by_department:
|
||||
yield from cls.render_for_department(department, certifications)
|
||||
|
||||
|
||||
def make_member_email(member, certifications):
|
||||
template = loader.get_template("paperwork/email/member_certifications.dj.html")
|
||||
class MemberCertificationEmail(CertificationEmailBase):
|
||||
template = "paperwork/email/member_certifications.dj.html"
|
||||
|
||||
html_body = template.render({"member": member, "certifications": certifications})
|
||||
@property
|
||||
def subject(self) -> str:
|
||||
return f"You have been issued {len(self.context['certifications'])} new CMS Certifications"
|
||||
|
||||
return make_multipart_email(
|
||||
f"You have been issued {len(certifications)} new CMS Certifications",
|
||||
html_body,
|
||||
to=[sanitize_address((member.account_name, member.email), "ascii")],
|
||||
)
|
||||
@classmethod
|
||||
def render_for_member(
|
||||
cls, member: Member, certifications: list[Certification]
|
||||
) -> mail.EmailMessage:
|
||||
return cls.render(
|
||||
{"member": member, "certifications": certifications},
|
||||
to=[sanitize_address((member.account_name, member.email), "ascii")],
|
||||
)
|
||||
|
||||
@classmethod
|
||||
def render_for_certifications(
|
||||
cls, ordered_certifications: QuerySet[Certification]
|
||||
) -> Iterable[mail.EmailMessage]:
|
||||
certifications_by_member = groupby(
|
||||
ordered_certifications.filter(member__isnull=False), lambda c: c.member
|
||||
)
|
||||
|
||||
for member, certifications in certifications_by_member:
|
||||
yield cls.render_for_member(member, list(certifications))
|
||||
|
||||
|
||||
def member_emails(ordered_queryset):
|
||||
certifications_by_member = groupby(
|
||||
ordered_queryset.filter(member__isnull=False), lambda c: c.member
|
||||
)
|
||||
class AdminCertificationEmail(CertificationEmailBase):
|
||||
template = "paperwork/email/admin_certifications.dj.html"
|
||||
|
||||
for member, certifications in certifications_by_member:
|
||||
yield make_member_email(member, list(certifications))
|
||||
@property
|
||||
def subject(self) -> str:
|
||||
return f"{len(self.context['certifications'])} new CMS Certifications issued"
|
||||
|
||||
|
||||
def admin_email(ordered_queryset):
|
||||
template = loader.get_template("paperwork/email/admin_certifications.dj.html")
|
||||
html_body = template.render({"certifications": ordered_queryset})
|
||||
|
||||
return make_multipart_email(
|
||||
f"{len(ordered_queryset)} new CMS Certifications issued",
|
||||
html_body,
|
||||
to=(
|
||||
get_user_model()
|
||||
@classmethod
|
||||
def render_for_certifications(
|
||||
cls, ordered_certifications: QuerySet[Certification]
|
||||
) -> Iterable[mail.EmailMessage]:
|
||||
yield cls.render(
|
||||
{"certifications": ordered_certifications},
|
||||
to=get_user_model()
|
||||
.objects.with_perm(
|
||||
"paperwork.receive_certification_emails",
|
||||
include_superusers=False,
|
||||
@ -95,16 +116,18 @@ def admin_email(ordered_queryset):
|
||||
# TODO: LDAPBackend does not support with_perm() directly
|
||||
backend="django.contrib.auth.backends.ModelBackend",
|
||||
)
|
||||
.values_list("email", flat=True)
|
||||
),
|
||||
)
|
||||
.values_list("email", flat=True),
|
||||
)
|
||||
|
||||
|
||||
def all_certification_emails(queryset):
|
||||
def all_certification_emails(queryset: QuerySet[Certification]):
|
||||
ordered_queryset = queryset.select_related(
|
||||
"certification_version__definition"
|
||||
).order_by("certification_version__definition__department")
|
||||
|
||||
yield from department_emails(ordered_queryset)
|
||||
yield from member_emails(ordered_queryset)
|
||||
yield admin_email(ordered_queryset)
|
||||
for email_type in [
|
||||
DepartmentCertificationEmail,
|
||||
MemberCertificationEmail,
|
||||
AdminCertificationEmail,
|
||||
]:
|
||||
yield from email_type.render_for_certifications(ordered_queryset)
|
||||
|
||||
Loading…
Reference in New Issue
Block a user