"""Transactional email delivery. Provider: Postmark. Picked for its transactional-deliverability reputation and a tiny, no-SDK-needed HTTP API. Configuration ------------- - ``POSTMARK_TOKEN`` / ``POSTMARK_TOKEN_FILE`` — server API token. - ``EMAIL_FROM`` — verified sender address (default ``licenses@datatools.unalogix.com``). - ``EMAIL_REPLY_TO`` — optional Reply-To (default same as From). When ``POSTMARK_TOKEN`` is unset the service falls back to :class:`LoggingEmailService`, which prints the email to stdout instead of sending. Lets the webhook handler exercise the full flow before the Postmark account is provisioned. """ from __future__ import annotations import logging import os from dataclasses import dataclass from typing import Optional, Protocol import httpx from app.config import get_settings log = logging.getLogger(__name__) @dataclass(frozen=True) class LicenseEmail: """Inputs the renderer needs from the caller.""" to_name: str to_email: str tier: str license_key: str expires_at_iso: str blob: str class EmailService(Protocol): """Provider-agnostic email surface — keeps Postmark out of the callers' import graph.""" def send_license(self, msg: LicenseEmail) -> str: """Deliver the license-delivery email. Returns a provider message id (or ``"logged"`` for the dev fallback) so the caller can record it on the licenses row for audit.""" ... class LoggingEmailService: """Stand-in when no real provider is configured. Logs the rendered message body at INFO so it shows up in ``docker compose logs api`` — useful during local dev and during the deploy window before Postmark is wired up.""" def send_license(self, msg: LicenseEmail) -> str: body = _render_text(msg) log.info( "[email-stub] would send to=%s subject=%r\n%s", msg.to_email, _subject(msg), body, ) return "logged" class PostmarkEmailService: """Postmark transactional API client. Single endpoint, ~3 fields, no SDK needed. We use a per-call httpx Client with a tight timeout — webhook handlers run on the request thread and we never want to block them on a flaky upstream. """ API_URL = "https://api.postmarkapp.com/email" TIMEOUT_S = 8.0 def __init__( self, token: str, *, sender: str, reply_to: Optional[str] = None, message_stream: str = "outbound", ) -> None: self._token = token self._sender = sender self._reply_to = reply_to or sender self._stream = message_stream def send_license(self, msg: LicenseEmail) -> str: body_text = _render_text(msg) body_html = _render_html(msg) payload = { "From": self._sender, "To": _rfc_addr(msg.to_name, msg.to_email), "ReplyTo": self._reply_to, "Subject": _subject(msg), "TextBody": body_text, "HtmlBody": body_html, "MessageStream": self._stream, } headers = { "Accept": "application/json", "Content-Type": "application/json", "X-Postmark-Server-Token": self._token, } with httpx.Client(timeout=self.TIMEOUT_S) as c: r = c.post(self.API_URL, json=payload, headers=headers) if r.status_code >= 400: raise EmailDeliveryError( f"Postmark rejected the request: HTTP {r.status_code} " f"body={r.text[:300]!r}" ) return str(r.json().get("MessageID", "")) class EmailDeliveryError(RuntimeError): """Provider returned a non-2xx. Caller should record this on the audit row so the operator can replay after fixing the provider config (verified sender domain, paid plan, etc.).""" # --------------------------------------------------------------------------- # Factory # --------------------------------------------------------------------------- def get_email_service() -> EmailService: """Choose the real provider if a token is configured, else the logger. Reads settings fresh — tests can flip env vars between sends without restarting.""" settings = get_settings() token = settings.resolve_postmark_token() if not token: return LoggingEmailService() sender = os.environ.get("EMAIL_FROM", "licenses@datatools.unalogix.com") reply_to = os.environ.get("EMAIL_REPLY_TO") return PostmarkEmailService(token, sender=sender, reply_to=reply_to) # --------------------------------------------------------------------------- # Rendering # --------------------------------------------------------------------------- def _subject(msg: LicenseEmail) -> str: return f"Your DataTools license ({msg.tier})" def _render_text(msg: LicenseEmail) -> str: return ( f"Hi {msg.to_name},\n\n" f"Thanks for your DataTools purchase. Your license is below.\n\n" f"License key: {msg.license_key}\n" f"Tier: {msg.tier}\n" f"Expires: {msg.expires_at_iso[:10]}\n\n" f"To activate, paste the full blob (starting with DTLIC2:) into\n" f"the Activate screen, or run:\n\n" f" python -m src.license_cli activate \"{msg.blob}\" \\\n" f" --name \"{msg.to_name}\" --email {msg.to_email}\n\n" f"Your blob:\n\n" f"{msg.blob}\n\n" f"Keep this email — you'll need the blob if you move to a new\n" f"computer. Questions: reply to this email.\n\n" f"— DataTools\n" ) def _render_html(msg: LicenseEmail) -> str: return ( "" f"

Hi {_html_escape(msg.to_name)},

" "

Thanks for your DataTools purchase. Your license is below.

" "" f"" f"" f"" "
License key{_html_escape(msg.license_key)}
Tier{_html_escape(msg.tier)}
Expires{_html_escape(msg.expires_at_iso[:10])}
" "

To activate, paste the blob below into the Activate " "screen on first launch.

" "
"
        f"{_html_escape(msg.blob)}
" "

Keep this email — you'll " "need the blob if you move to a new computer. Questions: just reply.

" "

— DataTools

" ) def _rfc_addr(name: str, email: str) -> str: # Postmark accepts "Name " or just "addr". Quote names with # special chars; otherwise keep it readable in the inbox. if not name or "@" in name: return email if any(c in name for c in ',<>"'): name = name.replace('"', "").replace(",", "") return f"{name} <{email}>" def _html_escape(s: str) -> str: return ( s.replace("&", "&") .replace("<", "<") .replace(">", ">") .replace('"', """) )