Email templating engine

.python-version

@@ -1 +1 @@
-3.12.9
+3.13.8

app/core/config.py

@@ -23,7 +23,9 @@ class Settings(BaseSettings):
         frozen=True,
     )
     # APP
-    RUN_ENV: Literal["local", "develop", "staging", "production"] = "local"
+    RUN_ENV: Literal["local", "develop", "staging", "production", "test"] = (
+        "local"
+    )
     PROCESS_TYPE: Literal["api", "worker", "beat"]
     API_V1_STR: str = "/api/v1"
     SECRET_KEY: str = secrets.token_urlsafe(32)

app/emails/services/emails_service.py

@@ -1,54 +1,34 @@
-from enum import Enum
-from string import Template
-
+from app import templates
 from app.core.config import settings
 from app.users.schemas.user_schema import UserInDB
+
+from app.emails._global_state import get_client
 from app.emails.clients.base import BaseEmailClient
 from app.emails.schema.email import Email, EmailContext
-from app.emails._global_state import get_client
-
-
-class Paths(Enum):
-    NEW_USER = "app/emails/templates/welcome_email.html"
 
 
 class EmailService:
     def __init__(self, email_client: BaseEmailClient | None = None):
         self.email_client = email_client or get_client()
-
-    def _get_email(
-        self,
-        recipient_email: str,
-        template: str,
-        subject: str,
-        html_message_input: dict | None = None,
-    ) -> Email:
-        with open(template, "r") as file:
-            html_template_string = file.read()
-
-        html_message_input = html_message_input or {}
-        html = Template(html_template_string).substitute(**html_message_input)
-
-        return Email(
-            to_emails=[recipient_email],
-            subject=subject,
-            html=html,
-        )
+        self.template_service = templates.TemplatesService()
 
     def send_new_user_email(
         self,
         user: UserInDB,
     ) -> None:
-        email = self._get_email(
-            user.email,
-            Paths.NEW_USER.value,
-            "Welcome",
-        )
-
-        email.context = EmailContext(
-            max_retries=settings.SEND_WELCOME_EMAIL_MAX_RETRIES,
-            backoff_in_seconds=settings.SEND_WELCOME_EMAIL_RETRY_BACKOFF_VALUE,
-            error_message=f"Sending new user email to user {user.id} failed",
+        from app.users.schemas.templates import NewUserTemplate
+
+        template = NewUserTemplate(name=user.email)
+
+        email = Email(
+            to_emails=[user.email],
+            subject="Welcome",
+            html=self.template_service.render(template),
+            context=EmailContext(
+                max_retries=settings.SEND_WELCOME_EMAIL_MAX_RETRIES,
+                backoff_in_seconds=settings.SEND_WELCOME_EMAIL_RETRY_BACKOFF_VALUE,
+                error_message=f"Sending new user email to user {user.id} failed",
+            ),
         )
 
         self.email_client.send_email(email)
@@ -57,10 +37,14 @@ def send_user_remind_email(
         self,
         user: UserInDB,
     ) -> None:
-        email = self._get_email(
-            user.email,
-            Paths.NEW_USER.value,
-            "Welcome",
+        from app.users.schemas.templates import NewUserTemplate
+
+        template = NewUserTemplate(name=user.email)
+
+        email = Email(
+            to_emails=[user.email],
+            subject="Welcome",
+            html=self.template_service.render(template),
         )
 
         self.email_client.send_email(email)

app/templates/__init__.py

@@ -0,0 +1,6 @@
+from . import types
+from . import exceptions
+from . import utils
+
+from .schemas import BaseTemplate, BaseEmailTemplate
+from .services import TemplatesService

app/templates/exceptions/__init__.py

@@ -0,0 +1,2 @@
+from .invalid_template_path_exception import InvalidTemplatePathException
+from .template_missing_path_exception import TemplateMissingPathException

app/templates/exceptions/invalid_template_path_exception.py

@@ -0,0 +1,12 @@
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from app.templates import BaseTemplate
+
+
+class InvalidTemplatePathException(Exception):
+    def __init__(self, obj: "BaseTemplate") -> None:
+        cls_name = obj.__class__.__name__
+        path = obj.get_path()
+        self.message = msg = f"{cls_name}: couldn't find template at {path}."
+        super().__init__(msg)

app/templates/exceptions/template_missing_path_exception.py

@@ -0,0 +1,25 @@
+from textwrap import dedent
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from app.templates import BaseTemplate
+
+
+class TemplateMissingPathException(Exception):
+    def __init__(self, obj: "BaseTemplate") -> None:
+        cls_name = obj.__class__.__name__
+        self.message = message = dedent(
+            f"""
+            {cls_name} does not have a path configured.
+
+            Set the path to the template file in the class definition, when
+            inheriting from BaseTemplate or BaseEmailTemplate.
+
+            Example:
+            ```
+            class {cls_name}(BaseTemplate, path="path/to/template.mj"):
+                ...
+            ```
+            """
+        )
+        super().__init__(message)

app/templates/schemas/__init__.py

@@ -0,0 +1,7 @@
+from .base_template import BaseTemplate
+from .base_email_template import BaseEmailTemplate
+
+__all__ = (
+    "BaseTemplate",
+    "BaseEmailTemplate",
+)

app/templates/schemas/base_email_template.py

@@ -0,0 +1,13 @@
+from abc import ABC
+
+from app.templates import utils
+
+from .base_template import BaseTemplate
+
+
+class BaseEmailTemplate(
+    BaseTemplate,
+    ABC,
+    pipeline=(utils.render_mjml,),
+):
+    pass

app/templates/schemas/base_template.py

@@ -0,0 +1,38 @@
+from abc import ABC
+from typing import Any, ClassVar
+
+import pydantic
+
+from app.templates import exceptions, types
+
+
+class BaseTemplate(pydantic.BaseModel, ABC):
+    __template_path__: ClassVar[str | None] = None
+    __template_pipeline__: ClassVar[types.ProcessingPipeline | None] = None
+
+    @classmethod
+    def __init_subclass__(
+        cls,
+        path: str | None = None,
+        pipeline: types.ProcessingPipeline | None = None,
+        **kw: Any,
+    ) -> None:
+        super().__init_subclass__(**kw)
+
+        if path:
+            cls.__template_path__ = path
+
+        if pipeline:
+            cls.__template_pipeline__ = pipeline
+
+    def get_path(self) -> str:
+        if self.__template_path__:
+            return self.__template_path__
+
+        raise exceptions.TemplateMissingPathException(self)
+
+    def get_pipeline(self) -> types.ProcessingPipeline:
+        return self.__template_pipeline__ or tuple()
+
+    def get_args(self) -> dict[str, Any]:
+        return self.model_dump(exclude="path")

app/templates/services/__init__.py

@@ -0,0 +1,5 @@
+from .templates_service import TemplatesService
+
+__all__ = (
+    "TemplatesService",
+)

app/templates/services/templates_service.py

@@ -0,0 +1,47 @@
+import itertools
+from typing import ClassVar
+
+import jinja2
+
+from app.templates import exceptions, schemas, types
+
+
+class TemplatesService:
+    ENVIRONMENT: ClassVar[jinja2.Environment] = jinja2.Environment(
+        loader=jinja2.FileSystemLoader("assets/templates/"),
+        trim_blocks=True,
+        lstrip_blocks=True,
+    )
+
+    def __init__(
+        self,
+        *,
+        environment: jinja2.Environment | None = None,
+        pipeline: types.ProcessingPipeline | None = None,
+    ) -> None:
+        self._environment = environment or self.ENVIRONMENT
+        self._pipeline = pipeline
+
+    def render(
+        self,
+        template: schemas.BaseTemplate,
+        *,
+        pipeline: types.ProcessingPipeline | None = None,
+    ) -> str:
+        path, args = template.get_path(), template.get_args()
+
+        try:
+            jinja_template = self._environment.get_template(path)
+        except jinja2.exceptions.TemplateNotFound:
+            raise exceptions.InvalidTemplatePathException(template)
+
+        rendered_template = jinja_template.render(**args)
+
+        for processor in itertools.chain(
+            template.get_pipeline(),
+            pipeline or [],
+            self._pipeline or [],
+        ):
+            rendered_template = processor(rendered_template)
+
+        return rendered_template

app/templates/types/__init__.py

@@ -0,0 +1,6 @@
+from .processor import Processor, ProcessingPipeline
+
+__all__ = (
+    "Processor",
+    "ProcessingPipeline",
+)

app/templates/types/processor.py

@@ -0,0 +1,4 @@
+from typing import Callable
+
+type Processor = Callable[[str], str]
+type ProcessingPipeline = tuple[Processor, ...]

app/templates/utils/__init__.py

@@ -0,0 +1,5 @@
+from .mjml_renderer import render_mjml
+
+__all__ = (
+    "render_mjml",
+)

app/templates/utils/mjml_renderer.py

@@ -0,0 +1,5 @@
+import mjml
+
+
+def render_mjml(input: str) -> str:
+    return mjml.mjml2html(input, disable_comments=True)

app/users/schemas/templates.py

@@ -0,0 +1,5 @@
+from app import templates
+
+
+class NewUserTemplate(templates.BaseEmailTemplate, path="new_user_email.j2"):
+    name: str

assets/templates/components/atoms/heading.j2

@@ -0,0 +1,31 @@
+{% from "constants/theme.j2" import theme %}
+{% from "utils/render_attributes.j2" import render_attributes %}
+
+{% macro text(variant='h1', styles={}) %}
+
+  {% set defaults = {
+    'color': theme.colors.primary,
+    'font-family': theme.typography.family,
+    'font-weight': theme.typography.weights.bold,
+    'line-height': theme.typography.line_heights.normal,
+    'padding-bottom': theme.spacing.lg,
+    'align': 'left',
+  } %}
+
+  {% set variants = {
+    'h1': {
+      'font-size': theme.typography.sizes.xxl,
+    },
+    'h2': {
+      'font-size': theme.typography.sizes.xl,
+    },
+    'h3': {
+      'font-size': theme.typography.sizes.lg,
+      'padding-bottom': theme.spacing.md,
+    },
+  } %}
+
+  <mj-text {{ render_attributes(defaults, variants.get(variant), styles) }}>
+    {{ caller() }}
+  </mj-text>
+{% endmacro %}

assets/templates/components/atoms/image.j2

@@ -0,0 +1,17 @@
+{% from "constants/theme.j2" import theme %}
+{% from "utils/render_attributes.j2" import render_attributes %}
+
+{% macro image(src, alt='', styles={}) %}
+  {% set defaults = {
+    'width': 'auto',
+    'align': 'center',
+    'padding-top': theme.spacing.lg,
+    'padding-bottom': theme.spacing.lg,
+  } %}
+
+  <mj-image
+    src="{{ src }}"
+    alt="{{ alt }}"
+    {{ render_attributes(defaults, styles) }}
+  />
+{% endmacro %}