"""Email utility classes and functions for rendering and sending templates."""
from __future__ import annotations
from dataclasses import dataclass
from types import MappingProxyType
from typing import TYPE_CHECKING, Any, ClassVar, cast
from django.core.mail import EmailMultiAlternatives, get_connection
from django.template.exceptions import TemplateDoesNotExist
from django.template.loader import render_to_string
from django.utils.html import strip_tags
from django.utils.translation import gettext_lazy as _
if TYPE_CHECKING:
from collections.abc import Iterable, Mapping
[docs]
Attachment = tuple[str, bytes, str] # (filename, content, mimetype)
@dataclass(frozen=True)
[docs]
class MailTemplate:
"""Represents a template base path with helpers for .txt and .html variants."""
[docs]
key: str # name of template (e.g. "user_welcome")
[docs]
base: str # template base path
[docs]
label: str # label for UI
[docs]
def txt(self) -> str:
"""Return the plain-text template path."""
return f'{self.base}.txt'
[docs]
def html(self) -> str:
"""Return the HTML template path."""
return f'{self.base}.html'
[docs]
class MailTemplates:
"""Registry of grouped mail templates."""
# --- User ---
[docs]
USER_WELCOME: ClassVar[MailTemplate] = MailTemplate(
key='user_welcome',
base='emails/user/user_welcome',
label=cast('str', _('User Welcome')),
)
[docs]
USER_DELETE: ClassVar[MailTemplate] = MailTemplate(
key='user_delete',
base='emails/user/user_delete',
label=cast('str', _('User Delete')),
)
# --- Certificate ---
[docs]
CERTIFICATE_ISSUED: ClassVar[MailTemplate] = MailTemplate(
key='certificate_issued',
base='emails/certificate/certificate_issued',
label=cast('str', _('Certificate Issued')),
)
[docs]
CERTIFICATE_REVOKED: ClassVar[MailTemplate] = MailTemplate(
key='certificate_revoked',
base='emails/certificate/certificate_revoked',
label=cast('str', _('Certificate Revoked')),
)
# Groups/registry
[docs]
GROUPS: ClassVar[Mapping[str, tuple[MailTemplate, ...]]] = {
'user': (USER_WELCOME, USER_DELETE),
'certificate': (CERTIFICATE_ISSUED, CERTIFICATE_REVOKED),
}
@classmethod
[docs]
def get_user_templates(cls) -> list[MailTemplate]:
"""Return the list of user-related templates."""
return list(cls.GROUPS['user'])
@classmethod
[docs]
def get_certificate_templates(cls) -> list[MailTemplate]:
"""Return the list of certificate-related templates."""
return list(cls.GROUPS['certificate'])
@classmethod
[docs]
def all(cls) -> list[MailTemplate]:
"""Return all templates for all groups."""
return [t for group in cls.GROUPS.values() for t in group]
@dataclass(frozen=True)
[docs]
class EmailPayload:
"""Immutable value object describing one outbound email.
Attributes:
subject: Subject line.
to: Recipients.
template_html: Django template path for the HTML body.
context: Template context (wrapped as read-only).
from_email: Optional override; defaults to settings.DEFAULT_FROM_EMAIL.
reply_to: Optional Reply-To addresses.
cc: Optional CC recipients.
bcc: Optional BCC recipients.
attachments: Optional sequence of (filename, bytes, mimetype).
headers: Optional extra headers (e.g., {"X-Tag": "welcome"}).
"""
[docs]
template_html: MailTemplate
[docs]
context: Mapping[str, object] = MappingProxyType({})
[docs]
from_email: str | None = None
[docs]
reply_to: tuple[str, ...] = ()
[docs]
cc: tuple[str, ...] = ()
[docs]
bcc: tuple[str, ...] = ()
[docs]
attachments: tuple[Attachment, ...] = ()
[docs]
def __post_init__(self) -> None:
"""Validate and normalize fields after dataclass initialization.
Ensures that mapping/sequence attributes are immutable wrappers or tuples.
"""
if not isinstance(self.context, MappingProxyType):
object.__setattr__(self, 'context', MappingProxyType(dict(self.context)))
if not isinstance(self.headers, MappingProxyType):
object.__setattr__(self, 'headers', MappingProxyType(dict(self.headers)))
if not isinstance(self.to, tuple):
object.__setattr__(self, 'to', tuple(self.to))
if not isinstance(self.reply_to, tuple):
object.__setattr__(self, 'reply_to', tuple(self.reply_to))
if not isinstance(self.cc, tuple):
object.__setattr__(self, 'cc', tuple(self.cc))
if not isinstance(self.bcc, tuple):
object.__setattr__(self, 'bcc', tuple(self.bcc))
if not isinstance(self.attachments, tuple):
object.__setattr__(self, 'attachments', tuple(self.attachments))
[docs]
def _render_bodies(tpl: MailTemplate, context: Mapping[str, object]) -> tuple[str, str]:
ctx = dict(context)
try:
text = render_to_string(tpl.txt(), ctx)
except TemplateDoesNotExist:
html = render_to_string(tpl.html(), ctx)
return strip_tags(html), html
else:
html = render_to_string(tpl.html(), ctx)
return text, html
[docs]
def send_email(payload: EmailPayload, *, connection: Any = None) -> int:
"""Send a single email with HTML + text alternative.
Returns:
Number of successfully delivered messages (0 or 1).
"""
text, html = _render_bodies(payload.template_html, payload.context)
msg = EmailMultiAlternatives(
subject=payload.subject,
body=text,
from_email=payload.from_email,
to=list(payload.to),
cc=list(payload.cc),
bcc=list(payload.bcc),
headers=dict(payload.headers),
reply_to=list(payload.reply_to),
connection=connection,
)
msg.attach_alternative(html, 'text/html')
for name, content, mimetype in payload.attachments:
msg.attach(name, content, mimetype)
return msg.send()
[docs]
def send_bulk(payloads: Iterable[EmailPayload]) -> int:
"""Send multiple emails reusing one SMTP connection."""
sent = 0
with get_connection() as conn:
for p in payloads:
sent += send_email(p, connection=conn)
return sent