Coverage for functions \ flipdare \ mailer \ _jinja_email_template.py: 82%
128 statements
« prev ^ index » next coverage.py v7.13.0, created at 2026-05-08 12:22 +1000
« prev ^ index » next coverage.py v7.13.0, created at 2026-05-08 12:22 +1000
1#!/usr/bin/env python
2# Copyright (c) 2026 Flipdare Pty Ltd. All rights reserved.
3#
4# This file is part of Flipdare's proprietary software and contains
5# confidential and copyrighted material. Unauthorised copying,
6# modification, distribution, or use of this file is strictly
7# prohibited without prior written permission from Flipdare Pty Ltd.
8#
9# This software includes third-party components licensed under MIT,
10# BSD, and Apache 2.0 licences. See THIRD_PARTY_NOTICES for details.
11#
14from abc import ABC, abstractmethod
15from pathlib import Path
16from typing import Any, cast
17from jinja2 import Environment, FileSystemLoader
18from pydantic import TypeAdapter, ValidationError
19from flipdare.app_log import LOG
20from flipdare.app_types import JinjaDataType
21from flipdare.constants import IS_TRACE
22from flipdare.mailer.app_email_params import AppEmailParams
23from flipdare.mailer.email_image import EmailImage
24from flipdare.error.app_error import ServerError
25from flipdare.error.data_load_error import DataLoadError
26from flipdare.generated.shared.app_error_code import AppErrorCode
28__all__ = ["JinjaEmailTemplate"]
31class JinjaEmailTemplate[TSchema](ABC):
32 __slots__ = (
33 "_data",
34 "_images",
35 "_jinja_env",
36 "_params",
37 "_text_data",
38 )
40 # NOTE: Subclasses MUST set this
41 SCHEMA_CLASS: type[TSchema] # pylint: disable=declare-non-slot
43 def __init__(
44 self,
45 data: TSchema | list[TSchema],
46 params: AppEmailParams[Any],
47 images: list[EmailImage] | None = None,
48 ) -> None:
49 # Setup Jinja2 environment
50 self._data = data
51 self._params = params
52 self._images = images
54 template_dir = self._params.abs_template_dir
55 try:
56 self._jinja_env = Environment(
57 loader=FileSystemLoader(template_dir),
58 autoescape=False, # We control the content # noqa: S701
59 trim_blocks=True,
60 lstrip_blocks=True,
61 )
62 except Exception as e:
63 cause = (
64 f"Error initializing Jinja2 environment\n"
65 f"\tCheck {template_dir} exists and is accessible.\n"
66 f"\tError: {e}"
67 )
69 LOG().error(cause)
70 raise ServerError(message=cause, error_code=AppErrorCode.EMAIL_TEMPLATE) from e
72 @abstractmethod
73 def newline_fields(self) -> list[str]: ...
75 @property
76 def data(self) -> TSchema | list[TSchema]:
77 return self._data
79 @property
80 def subject(self) -> str:
81 return self._params.subject
83 @property
84 def images(self) -> list[EmailImage] | None:
85 return self._images
87 @property
88 def jinja_env(self) -> Environment:
89 return self._jinja_env
91 @property
92 def html(self) -> str:
93 """Render the email as HTML using Jinja2 templates."""
94 return self.render_html()
96 @property
97 def text(self) -> str:
98 """Render the email as plain text using Jinja2 templates."""
99 return self.render_text()
101 @property
102 def html_name(self) -> str:
103 return self._params.html_name
105 @property
106 def text_name(self) -> str:
107 return self._params.text_name
109 @property
110 def html_path(self) -> Path:
111 file_path = self._params.html_path
113 if not file_path.exists():
114 msg = f"Template path ({file_path}) not found."
115 raise ServerError(message=msg, error_code=AppErrorCode.EMAIL_TEMPLATE)
117 if IS_TRACE:
118 LOG().trace(f"Using template path: {file_path}")
120 return file_path
122 @property
123 def text_path(self) -> Path:
124 file_path = self._params.text_path
126 if not file_path.exists():
127 msg = f"Template path ({file_path}) not found."
128 raise ServerError(message=msg, error_code=AppErrorCode.EMAIL_TEMPLATE)
130 if IS_TRACE:
131 LOG().trace(f"Using template path: {file_path}")
133 return file_path
135 @property
136 def debug_label(self) -> str:
137 return f"Template={self.subject}, HTML={self.html_name}, Text={self.text_name}"
139 def validate(self) -> None:
140 from flipdare.error.message_format import JinjaTmplErrorMsgFormat
142 try:
143 self.validate_data()
144 except Exception as e:
145 formatter = JinjaTmplErrorMsgFormat(
146 template_name=self.debug_label,
147 error=e,
148 jinja_label=self.debug_label,
149 )
151 LOG().error(str(formatter))
152 raise ServerError(
153 message=str(formatter),
154 error_code=AppErrorCode.EMAIL_TEMPLATE,
155 ) from e
157 def render_html(self) -> str:
158 """Render the daily summary as HTML using Jinja2 templates."""
159 return self._render(name=self.html_name, data=self._parse_for_html())
161 def render_text(self) -> str:
162 """Render the daily summary as plain text using Jinja2 templates."""
163 return self._render(name=self.text_name, data=self._parse_for_text())
165 def _render(self, name: str, data: JinjaDataType) -> str:
166 template = self._jinja_env.get_template(name)
167 match data:
168 case list():
169 return str(template.render(sections=data))
170 case dict():
171 return str(template.render(**data))
173 def _parse_for_text(self) -> JinjaDataType:
174 return self.to_dict()
176 def _parse_for_html(self) -> JinjaDataType:
177 parsed_data = self.to_dict()
179 newline_fields = self.newline_fields()
180 if len(newline_fields) == 0:
181 return parsed_data
183 assert isinstance(parsed_data, dict) # narrowing, we known we have a dict..
184 return self._update_newline_fields(data=parsed_data, fields=newline_fields)
186 def to_dict(self) -> JinjaDataType:
187 validated = self.validate_data()
188 if isinstance(validated, list):
189 return [cast("dict[str, Any]", item) for item in validated]
190 else:
191 return cast("dict[str, Any]", validated)
193 def validate_data(self) -> TSchema | list[TSchema]:
194 from flipdare.error.message_format import ValidationErrorMsgFormat
196 # Pydantic's TypeAdapter is strict about the type it's initialized with.
197 # If self.SCHEMA_CLASS is SummaryEmailSchema, it expects the input to be a dict.
198 target_type = self.SCHEMA_CLASS
199 data = self._data
200 if isinstance(data, list):
201 target_type = list[self.SCHEMA_CLASS] # type: ignore
203 # Validate with Pydantic
204 try:
205 adapter = TypeAdapter(target_type)
206 return adapter.validate_python(data)
208 except ValidationError as e:
209 class_name = self.SCHEMA_CLASS.__name__
210 formatter = ValidationErrorMsgFormat(
211 class_type=self.SCHEMA_CLASS,
212 error=e,
213 )
214 LOG().error(str(formatter))
215 raise DataLoadError.model(
216 class_name=class_name,
217 missing_code=formatter.user_error_code,
218 error=e,
219 ) from e
221 def _update_newline_fields(self, data: dict[str, Any], fields: list[str]) -> dict[str, Any]:
222 html_data = data.copy()
223 for field in fields:
224 if field not in data:
225 continue
227 old_value = data[field]
228 html_data[field] = old_value.replace("\n", "<br>")
230 return html_data