"""Caller-side Email Template consumption. Not a second renderer and not email.send. Composition declares which catalog key and how to bind vars. Cloud render remains the only interpolation/escaping/missing-variable implementation. Callers then pass concrete RenderedEmail fields into email.send. """ from __future__ import annotations from collections.abc import Mapping from typing import Any, Literal from pydantic import BaseModel, ConfigDict, Field, StrictStr, model_validator from src.domain.codegen.email_templates_manifest import ( EMAIL_TEMPLATE_CATALOG, declared_variable_names, ) from src.domain.connectors.schemas import InvokeEmailSendRequest VariableSource = Literal["literal", "field"] UNKNOWN_TEMPLATE_KEY = "unknown_template_key" UNKNOWN_TEMPLATE_VARIABLE = "unknown_template_variable" MISSING_REQUIRED_VARIABLE_BINDING = "missing_required_variable_binding" INVALID_VARIABLE_BINDING = "invalid_variable_binding" SOURCE_FIELD_MISSING = "source_field_missing" INVALID_FIELD_VALUE = "invalid_field_value" MISSING_REQUIRED_VALUE = "missing_required_value" EMAIL_TEMPLATE_CONSUMPTION: dict[str, Any] = { "slice_id": "8f_email_template_composition_consumption", "status": "complete", "template_key_on_email_send": False, "interpolation": False, "new_email_verb": False, "forms": "additive", "render_owner": "cloud", "addressing_owner": "caller", "variable_sources": ("literal", "field"), "docs": "docs/audits/EMAIL_TEMPLATES_DOMAIN_8F.md", } class EmailTemplateVariableBinding(BaseModel): """Maps one catalog variable to a literal or a context field. No expressions.""" model_config = ConfigDict(extra="forbid", frozen=True, strict=True) source: VariableSource value: StrictStr | None = None field: StrictStr | None = None @model_validator(mode="after") def source_shape(self): if self.source == "literal": if self.value is None or self.field is not None: raise ValueError(INVALID_VARIABLE_BINDING) elif self.field is None or not self.field.strip() or self.value is not None: raise ValueError(INVALID_VARIABLE_BINDING) return self class EmailTemplateInvocation(BaseModel): """Caller/composition resource. Not an email.send payload and not RenderedEmail.""" model_config = ConfigDict(extra="forbid", frozen=True, strict=True) template_key: StrictStr = Field(min_length=1) variables: dict[str, EmailTemplateVariableBinding] = Field(default_factory=dict) @model_validator(mode="after") def catalog_contract(self): catalog_entry = EMAIL_TEMPLATE_CATALOG.get(self.template_key) if catalog_entry is None: raise ValueError(UNKNOWN_TEMPLATE_KEY) allowed = declared_variable_names(catalog_entry) unknown = sorted(name for name in self.variables if name not in allowed) if unknown: raise ValueError(f"{UNKNOWN_TEMPLATE_VARIABLE}:{unknown}") missing = sorted( name for name in (catalog_entry.get("required_variables") or []) if name not in self.variables ) if missing: raise ValueError(f"{MISSING_REQUIRED_VARIABLE_BINDING}:{missing}") return self class EmailTemplateVarResolution(BaseModel): model_config = ConfigDict(extra="forbid", frozen=True) ok: bool vars: dict[str, str] = Field(default_factory=dict) error_code: StrictStr | None = None template_key: StrictStr variable: StrictStr | None = None def resolve_email_template_variables( invocation: EmailTemplateInvocation, context: Mapping[str, Any], ) -> EmailTemplateVarResolution: """Build the string map for Cloud render. Does not interpolate or escape.""" catalog_entry = EMAIL_TEMPLATE_CATALOG[invocation.template_key] required = set(catalog_entry.get("required_variables") or []) resolved: dict[str, str] = {} for name, binding in invocation.variables.items(): if binding.source == "literal": value = binding.value or "" else: if binding.field not in context: if name in required: return EmailTemplateVarResolution( ok=False, error_code=SOURCE_FIELD_MISSING, template_key=invocation.template_key, variable=name, ) continue raw = context[binding.field] if raw is None: if name in required: return EmailTemplateVarResolution( ok=False, error_code=SOURCE_FIELD_MISSING, template_key=invocation.template_key, variable=name, ) continue if not isinstance(raw, str): return EmailTemplateVarResolution( ok=False, error_code=INVALID_FIELD_VALUE, template_key=invocation.template_key, variable=name, ) value = raw if name in required and value == "": return EmailTemplateVarResolution( ok=False, error_code=MISSING_REQUIRED_VALUE, template_key=invocation.template_key, variable=name, ) if name not in required and value == "": continue resolved[name] = value missing_required = sorted(name for name in required if name not in resolved or resolved[name] == "") if missing_required: return EmailTemplateVarResolution( ok=False, error_code=MISSING_REQUIRED_VALUE, template_key=invocation.template_key, variable=missing_required[0], ) return EmailTemplateVarResolution(ok=True, vars=resolved, template_key=invocation.template_key) def concrete_email_send_payload( *, to: str | list[str], rendered: Mapping[str, Any], reply_to: str | None = None, ) -> dict[str, Any]: """Handoff from RenderedEmail to email.send. Drops template metadata.""" payload: dict[str, Any] = { "to": to, "subject": rendered["subject"], "text": rendered["text"], } html = rendered.get("html") if html: payload["html"] = html if reply_to: payload["reply_to"] = reply_to dumped = InvokeEmailSendRequest.model_validate(payload).model_dump(exclude_none=True) assert "template_key" not in dumped assert "vars" not in dumped return dumped CONTACT_CAPTURE_ADMIN_NOTIFY = EmailTemplateInvocation( template_key="contact_capture.admin_notify", variables={ "name": EmailTemplateVariableBinding(source="field", field="name"), "email": EmailTemplateVariableBinding(source="field", field="email"), "message": EmailTemplateVariableBinding(source="field", field="message"), }, ) CONTACT_CAPTURE_CONFIRMATION = EmailTemplateInvocation( template_key="contact_capture.confirmation", variables={ "name": EmailTemplateVariableBinding(source="field", field="name"), }, ) FEATURE_PACK_EMAIL_TEMPLATE_CONSUMPTION: dict[str, tuple[EmailTemplateInvocation, ...]] = { "contact-capture": (CONTACT_CAPTURE_ADMIN_NOTIFY, CONTACT_CAPTURE_CONFIRMATION), } APP_BLOCK_EMAIL_TEMPLATE_POLICY: dict[str, Any] = { "forms.contact": { "requires": ("email.send",), "template_keys": (), "note": "delivery capability only; pack contact-capture owns template keys", }, "forms.quote_request": { "requires": ("email.send",), "template_keys": (), "note": "delivery capability only", }, } SURFACE_CLASSIFICATION: dict[str, str] = { "auth_cloud_in_process": "compatible", "contact_capture_notify_function": "supported", "email_template_invocation": "canonical", "feature_pack_contact_capture": "supported", "app_block_forms_contact": "canonical", "forms_7j": "unchanged", "forms_template_mode": "canonical", "legacy_api_contact": "compatibility_only", "legacy_api_contact_generation": "deprecated_for_new_projects", "generated_app_local_renderer": "out_of_scope", }