Coverage for src / lexigram / contracts / mailer / types.py: 100%
73 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-08-19 05:41 +0800
1# lexigram-contracts/src/lexigram/contracts/mailer/types.py
2"""Mailer value types."""
4from __future__ import annotations
6from dataclasses import dataclass, field
7from enum import StrEnum
8import re
9from typing import TYPE_CHECKING
11from lexigram.contracts.lib.time import utcnow as _utcnow
13if TYPE_CHECKING:
14 from datetime import datetime
16_HEADER_NAME_RE = re.compile(r"[!-9;-~]+")
17"""RFC 5322 field-name token: printable ASCII excluding space and colon."""
20def _reject_crlf(value: str | None, field: str) -> None:
21 """Reject CR/LF in a field that will be serialized into a MIME header.
23 Args:
24 value: Field value to check; ``None`` passes.
25 field: Human-readable field path for the error message.
27 Raises:
28 ValueError: If the value contains a CR or LF character.
29 """
30 if value is not None and ("\r" in value or "\n" in value):
31 raise ValueError(f"{field} must not contain CR or LF characters")
34@dataclass(frozen=True)
35class EmailMessage:
36 """A cross-channel email message value object.
38 Passed to :class:`~lexigram.contracts.mailer.protocols.MailerProtocol`.
39 All fields except *to* and *subject* are optional to keep simple use
40 cases concise.
42 Attributes:
43 to: List of recipient email addresses.
44 subject: Email subject line.
45 body: Plain-text message body.
46 html_body: HTML message body; falls back to *body* when absent.
47 from_email: Sender address; implementations should supply a
48 safe default when ``None``.
49 from_name: Human-readable sender name.
50 reply_to: Optional reply-to address.
51 cc: Carbon-copy recipients.
52 bcc: Blind carbon-copy recipients.
53 headers: Additional MIME headers.
54 """
56 to: list[str]
57 subject: str
58 body: str = ""
59 html_body: str | None = None
60 from_email: str | None = None
61 from_name: str | None = None
62 reply_to: str | None = None
63 cc: list[str] = field(default_factory=list)
64 bcc: list[str] = field(default_factory=list)
65 headers: dict[str, str] = field(default_factory=dict)
67 def __post_init__(self) -> None:
68 """Reject CR/LF in header-bound fields to prevent SMTP header injection.
70 Only fields that are serialized into MIME headers are constrained;
71 ``body`` and ``html_body`` are message content and may contain
72 newlines. Raise (fail closed) rather than silently stripping, so a
73 caller never believes an altered message was delivered as written.
75 Raises:
76 ValueError: If any header-bound field contains CR/LF or a
77 header name is not a valid RFC 5322 field-name token.
78 """
79 _reject_crlf(self.subject, "EmailMessage.subject")
80 _reject_crlf(self.from_email, "EmailMessage.from_email")
81 _reject_crlf(self.from_name, "EmailMessage.from_name")
82 _reject_crlf(self.reply_to, "EmailMessage.reply_to")
83 for recipient_field, recipients in (
84 ("to", self.to),
85 ("cc", self.cc),
86 ("bcc", self.bcc),
87 ):
88 for recipient in recipients:
89 _reject_crlf(recipient, f"EmailMessage.{recipient_field}")
90 for name, value in self.headers.items():
91 if not _HEADER_NAME_RE.fullmatch(name):
92 raise ValueError(
93 f"EmailMessage.headers contains invalid header name: {name!r}"
94 )
95 _reject_crlf(value, f"EmailMessage.headers[{name!r}]")
98@dataclass(frozen=True)
99class MessageDeliveryReceipt:
100 """Confirmation that a message was accepted by the delivery backend.
102 This is the success value returned by all messaging send operations.
103 It does NOT guarantee delivery to the end recipient; it only confirms
104 that the backend accepted the request.
106 Attributes:
107 message_id: Unique identifier for the message, assigned by the
108 framework (or the backend when ``provider_reference`` differs).
109 backend: Name of the backend that handled the send.
110 channel: Delivery channel: ``"email"``, ``"sms"``, or ``"push"``.
111 sent_at: UTC timestamp when the backend accepted the message.
112 provider_reference: Opaque reference returned by the external
113 provider. ``None`` when the backend does not return a reference.
114 """
116 message_id: str
117 backend: str
118 channel: str
119 sent_at: datetime = field(default_factory=_utcnow)
120 provider_reference: str | None = None
123class MessagePriority(StrEnum):
124 """Message delivery priority."""
126 LOW = "low"
127 NORMAL = "normal"
128 HIGH = "high"
129 URGENT = "urgent"
132class MessageStatus(StrEnum):
133 """Outbound message status."""
135 PENDING = "pending"
136 SENT = "sent"
137 FAILED = "failed"
138 CANCELLED = "cancelled"
141class DeliveryState(StrEnum):
142 """Downstream delivery state reported by provider."""
144 QUEUED = "queued"
145 DELIVERED = "delivered"
146 BOUNCED = "bounced"
147 REJECTED = "rejected"
148 DEFERRED = "deferred"
151@dataclass(frozen=True)
152class MessageAddress:
153 """An email address with an optional display name.
155 Attributes:
156 email: RFC-5321 email address.
157 name: Human-readable display name; omitted when ``None``.
158 """
160 email: str
161 name: str | None = None
163 def __str__(self) -> str:
164 if self.name:
165 return f"{self.name} <{self.email}>"
166 return self.email
169@dataclass(frozen=True)
170class Attachment:
171 """An email attachment.
173 Attributes:
174 filename: Suggested filename shown to the recipient.
175 content: Raw bytes of the attachment.
176 content_type: MIME type; defaults to ``application/octet-stream``.
177 content_id: Optional CID for inline embedding (e.g. ``cid:logo``).
178 """
180 filename: str
181 content: bytes
182 content_type: str = "application/octet-stream"
183 content_id: str | None = None
186__all__ = [
187 "Attachment",
188 "DeliveryState",
189 "EmailMessage",
190 "MessageAddress",
191 "MessageDeliveryReceipt",
192 "MessagePriority",
193 "MessageStatus",
194]