"""Email provider implementations (Resend, Mailgun, SES). Implements EmailProviderProtocol from core.protocols. Domain services depend on the protocol; this module is the only place that knows which transport is configured. """ from __future__ import annotations import asyncio from email.utils import formataddr, parseaddr from typing import Any, Callable, Optional import httpx import structlog from src.config import Settings, get_settings from src.core.protocols import SendResult logger = structlog.get_logger() SesClientFactory = Callable[..., Any] class ResendProvider: """Resend email provider.""" def __init__(self, api_key: Optional[str] = None, from_email: Optional[str] = None): settings = get_settings() self._api_key = api_key or settings.resend_api_key self._from = formataddr( (settings.from_name, parseaddr(from_email or settings.from_email)[1]) ) async def send( self, to: str, subject: str, html: str, text: Optional[str] = None, reply_to: Optional[str] = None, ) -> SendResult: payload = { "from": self._from, "to": [to], "subject": subject, "html": html, "text": text, } if reply_to: payload["reply_to"] = reply_to async with httpx.AsyncClient() as client: try: resp = await client.post( "https://api.resend.com/emails", headers={"Authorization": f"Bearer {self._api_key}"}, json=payload, ) resp.raise_for_status() data = resp.json() logger.info("email_sent_resend", to=to, message_id=data.get("id")) return SendResult(success=True, message_id=data.get("id")) except Exception as e: logger.error("email_send_failed_resend", to=to, error=str(e)) return SendResult(success=False, error=str(e)) class MailgunProvider: """Mailgun email provider.""" def __init__(self, api_key: Optional[str] = None, domain: Optional[str] = None, from_email: Optional[str] = None): settings = get_settings() self._api_key = api_key or settings.mailgun_api_key self._domain = domain or settings.mailgun_domain self._from = formataddr( (settings.from_name, parseaddr(from_email or settings.from_email)[1]) ) async def send( self, to: str, subject: str, html: str, text: Optional[str] = None, reply_to: Optional[str] = None, ) -> SendResult: data = { "from": self._from, "to": [to], "subject": subject, "html": html, "text": text or "", } if reply_to: data["h:Reply-To"] = reply_to async with httpx.AsyncClient() as client: try: resp = await client.post( f"https://api.mailgun.net/v3/{self._domain}/messages", auth=("api", self._api_key), data=data, ) resp.raise_for_status() data = resp.json() logger.info("email_sent_mailgun", to=to, message_id=data.get("id")) return SendResult(success=True, message_id=data.get("id")) except Exception as e: logger.error("email_send_failed_mailgun", to=to, error=str(e)) return SendResult(success=False, error=str(e)) class SesProvider: """AWS SES email provider. Uses boto3; keys are optional when IAM is available.""" def __init__( self, *, from_email: Optional[str] = None, region: Optional[str] = None, access_key_id: Optional[str] = None, secret_access_key: Optional[str] = None, configuration_set: Optional[str] = None, client_factory: Optional[SesClientFactory] = None, ): settings = get_settings() self._from = formataddr( (settings.from_name, parseaddr(from_email or settings.from_email)[1]) ) self._region = (region or settings.ses_region or "us-east-1").strip() key_id = access_key_id if access_key_id is not None else settings.ses_access_key_id secret = secret_access_key if secret_access_key is not None else settings.ses_secret_access_key self._access_key_id = key_id.strip() self._secret_access_key = secret.strip() self._configuration_set = ( configuration_set if configuration_set is not None else settings.ses_configuration_set ).strip() self._client_factory = client_factory def _client_kwargs(self) -> dict[str, str]: kwargs = {"region_name": self._region} if self._access_key_id and self._secret_access_key: kwargs["aws_access_key_id"] = self._access_key_id kwargs["aws_secret_access_key"] = self._secret_access_key return kwargs def _build_client(self): if self._client_factory is not None: return self._client_factory(**self._client_kwargs()) import boto3 return boto3.client("ses", **self._client_kwargs()) def _send_sync( self, to: str, subject: str, html: str, text: Optional[str], reply_to: Optional[str], ) -> dict[str, Any]: message: dict[str, Any] = { "Subject": {"Data": subject, "Charset": "UTF-8"}, "Body": {"Html": {"Data": html, "Charset": "UTF-8"}}, } if text: message["Body"]["Text"] = {"Data": text, "Charset": "UTF-8"} params: dict[str, Any] = { "Source": self._from, "Destination": {"ToAddresses": [to]}, "Message": message, } if reply_to: params["ReplyToAddresses"] = [reply_to] if self._configuration_set: params["ConfigurationSetName"] = self._configuration_set return self._build_client().send_email(**params) async def send( self, to: str, subject: str, html: str, text: Optional[str] = None, reply_to: Optional[str] = None, ) -> SendResult: try: data = await asyncio.to_thread(self._send_sync, to, subject, html, text, reply_to) message_id = data.get("MessageId") logger.info("email_sent_ses", to=to, message_id=message_id, region=self._region) return SendResult(success=True, message_id=message_id) except Exception as e: logger.error("email_send_failed_ses", to=to, error=str(e), region=self._region) return SendResult(success=False, error=str(e)) def normalized_email_provider(settings: Optional[Settings] = None) -> str: settings = settings or get_settings() return (settings.email_provider or "resend").strip().lower() or "resend" def email_transport_configured(settings: Optional[Settings] = None) -> bool: """True when the selected transport has enough config to attempt a send.""" settings = settings or get_settings() provider = normalized_email_provider(settings) if provider == "mailgun": return bool(settings.mailgun_api_key and settings.mailgun_domain) if provider == "ses": return bool((settings.from_email or "").strip()) return bool(settings.resend_api_key) def get_email_provider(from_email: Optional[str] = None): """Factory: return the configured email provider.""" settings = get_settings() provider = normalized_email_provider(settings) if provider == "mailgun": return MailgunProvider(from_email=from_email) if provider == "ses": return SesProvider(from_email=from_email) return ResendProvider(from_email=from_email) CLOUD_EMAIL_PROVIDERS = ("resend", "mailgun", "ses") def _cloud_provider_ready(settings: Settings, provider: str) -> bool: from_email = (settings.cloud_email_from or "").strip() if not from_email: return False if provider == "mailgun": return bool(settings.cloud_mailgun_api_key and settings.cloud_mailgun_domain) if provider == "ses": return bool(settings.cloud_ses_access_key_id and settings.cloud_ses_secret_access_key) if provider == "resend": return bool(settings.cloud_resend_api_key) return False def normalized_cloud_email_provider(settings: Settings | None = None) -> str: """Provider used for generated-app Cloud mail — never Core EMAIL_PROVIDER. If ``CLOUD_EMAIL_PROVIDER`` is set and that transport has credentials, it wins. Otherwise pick a transport that actually has ``CLOUD_*`` credentials so SES keys are used when Resend was left as the unused default. """ settings = settings or get_settings() requested = (settings.cloud_email_provider or "").strip().lower() if requested in CLOUD_EMAIL_PROVIDERS and _cloud_provider_ready(settings, requested): return requested for candidate in ("ses", "mailgun", "resend"): if _cloud_provider_ready(settings, candidate): return candidate return requested if requested in CLOUD_EMAIL_PROVIDERS else "resend" def _sanitize_from_display_name(name: str) -> str: cleaned = " ".join(name.replace('"', "").replace("<", "").replace(">", "").split()) return cleaned[:78] def cloud_email_from_header(settings: Settings | None = None, *, display_name: str | None = None) -> str: """From header for generated-app Cloud mail. Prefer the project's display name (e.g. ``RF Consultant``) so recipients see the customer app, not the platform. ``CLOUD_EMAIL_FROM_NAME`` is the fallback when the project has no name. The address itself stays ``CLOUD_EMAIL_FROM``. """ settings = settings or get_settings() email = (settings.cloud_email_from or "").strip() name = _sanitize_from_display_name((display_name or settings.cloud_email_from_name or "").strip()) if name and email: return formataddr((name, email)) return email def cloud_email_platform_footer(settings: Settings | None = None) -> tuple[str, str]: """Restrained attribution for platform-managed transactional delivery.""" _ = settings or get_settings() text = "Sent securely via P2HS" markup = ( '

' "Sent securely via P2HS" "

" ) return text, markup def apply_platform_email_branding(policy: Any, settings: Settings, *, project_name: str | None) -> None: policy.email_from = cloud_email_from_header(settings, display_name=project_name) reply_to = (settings.cloud_email_reply_to or "").strip() policy.email_reply_to = reply_to or None text, markup = cloud_email_platform_footer(settings) policy.email_footer_text = text policy.email_footer_html = markup def cloud_email_transport_configured(settings: Settings | None = None) -> bool: """True when the Cloud/generated-app default transport can send. Credentials are independent of EMAIL_PROVIDER / RESEND_API_KEY / MAILGUN_* / SES_*. Cloud Workers must be given the matching CLOUD_EMAIL_* secrets to actually deliver. """ settings = settings or get_settings() return _cloud_provider_ready(settings, normalized_cloud_email_provider(settings))