Middleware example.

Author: ianjosephwilson

tdom/cvar_utils.py

@@ -0,0 +1,29 @@
+from contextvars import ContextVar, Token
+
+
+class ContextVarSetter:
+    """
+    Context manager for working with many context vars (instead of only 1).
+
+    This is meant to be created, used immediately and then discarded.
+
+    This allows for dynamically specifying a tuple of var / value pairs that
+    another part of the program can use to wrap some called code without knowing
+    anything about either.
+    """
+
+    context_values: tuple[tuple[ContextVar, object], ...]  # Cvar / value pair.
+    tokens: tuple[Token, ...]
+
+    def __init__(self, context_values=()):
+        self.context_values = context_values
+        self.tokens = ()
+
+    def __enter__(self):
+        """Set every given context var to its paired value."""
+        self.tokens = tuple(var.set(val) for var, val in self.context_values)
+
+    def __exit__(self, exc_type, exc_value, traceback):
+        """Reset every given context var."""
+        for idx, var_value in enumerate(self.context_values):
+            var_value[0].reset(self.tokens[idx])

tdom/cvar_utils_test.py

@@ -0,0 +1,72 @@
+import string
+from contextvars import ContextVar
+
+from .cvar_utils import ContextVarSetter
+
+CtxStr = ContextVar[str]("CtxStr", default="default")
+CtxInt = ContextVar[int]("CtxInt", default=0)
+
+
+def _assert_ctx(ctx_str: str = "default", ctx_int: int = 0):
+    assert CtxStr.get() == ctx_str
+    assert CtxInt.get() == ctx_int
+
+
+def test_set():
+    _assert_ctx()
+    with ContextVarSetter(
+        context_values=(
+            (CtxStr, "new"),
+            (CtxInt, 1),
+        )
+    ):
+        _assert_ctx("new", 1)
+    _assert_ctx()
+
+
+def test_nest():
+    _assert_ctx()
+    with ContextVarSetter(
+        context_values=(
+            (CtxStr, "new"),
+            (CtxInt, 1),
+        )
+    ):
+        _assert_ctx("new", 1)
+        with ContextVarSetter(
+            context_values=(
+                (CtxStr, "again"),
+                (CtxInt, 2),
+            )
+        ):
+            _assert_ctx("again", 2)
+        _assert_ctx("new", 1)
+    _assert_ctx()
+
+
+def test_reps():
+    _assert_ctx()
+    for index, leter in enumerate(string.ascii_lowercase):
+        with ContextVarSetter(
+            context_values=(
+                (CtxStr, leter),
+                (CtxInt, index),
+            )
+        ):
+            _assert_ctx(leter, index)
+        _assert_ctx()
+
+
+def test_empty():
+    _assert_ctx()
+    with ContextVarSetter(context_values=()):
+        # DO NOTHING BUT NOT AN ERROR
+        _assert_ctx()
+    _assert_ctx()
+
+
+def test_one():
+    _assert_ctx()
+    with ContextVarSetter(context_values=((CtxStr, "new"),)):
+        _assert_ctx("new")
+    _assert_ctx()

tdom/processor.py

@@ -1,12 +1,14 @@
 import typing as t
 from collections.abc import Callable, Iterable, Sequence
+from contextvars import ContextVar
 from dataclasses import dataclass, field
 from functools import lru_cache
 from string.templatelib import Interpolation, Template
 
 from markupsafe import Markup
 
 from .callables import CallableInfo, get_callable_info
+from .cvar_utils import ContextVarSetter
 from .escaping import (
     escape_html_comment as default_escape_html_comment,
 )
@@ -527,7 +529,7 @@ def process(
         attrs: tuple[TAttribute, ...],
         component_template: Template,
         provided_attrs: tuple[Attribute, ...] = (),
-    ) -> Template:
+    ) -> Template | tuple[Template, object]:
         """
         Process available component details into a Template.
         """
@@ -547,7 +549,7 @@ def process(
         attrs: tuple[TAttribute, ...],
         component_template: Template,
         provided_attrs: tuple[Attribute, ...] = (),
-    ) -> Template:
+    ) -> Template | tuple[Template, object]:
         """
         Process available component details into a Template.
 
@@ -585,21 +587,48 @@ def process(
         )
         res1 = component_callable(**kwargs)  # ty: ignore[call-top-callable]
         if isinstance(res1, Template):
-            return res1
+            return res1, None
         elif callable(res1):
             res2 = res1()  # ty: ignore[call-top-callable]
             if isinstance(res2, Template):
                 return res2
+            elif isinstance(res2, tuple):
+                if len(res2) == 2:
+                    if not isinstance(res2[0], Template):
+                        raise TypeError(
+                            f"Component object returned unxpected type in first entry of 2-tuple: {type(res2[0])}"
+                        )
+                    else:
+                        # @TYPING:
+                        # Rebuild tuple so TY can correctly narrow types,
+                        # pyright works with `return res2`.
+                        return (res2[0], res2[1])
+                else:
+                    raise ValueError(
+                        f"Component object returned tuple with length != 2: {len(res2)}"
+                    )
+
             else:
                 raise TypeError(
-                    f"Component object must return Template when called: {type(res2)}"
+                    f"Component object must return Template or 2-tuple when called: {type(res2)}"
                 )
         else:
             raise TypeError(
                 f"Component callable must return Template or Callable: {type(res1)}"
             )
 
 
+@t.runtime_checkable
+class IMiddlewareGetContextValues(t.Protocol):
+    """
+    Middleware that provides a tuple of 2-tuples each with a context variable
+    paired with a value to set when processing the component's template.
+    """
+
+    # @TODO: Can we match a contextvar's type with the value to set like this?
+    def get_context_values(self) -> tuple[tuple[ContextVar, object], ...]: ...
+
+
 class ITemplateProcessor(t.Protocol):
     def process(self, root_template: Template, assume_ctx: ProcessContext) -> str: ...
 
@@ -817,10 +846,44 @@ def _process_component(
             template, attrs, start_i_index, end_i_index, check_callables=True
         )
         component_callable = template.interpolations[start_i_index].value
-        result_t = self.component_processor_api.process(
+        result = self.component_processor_api.process(
             template, last_ctx, component_callable, attrs, children_template
         )
-        return self._process_template(result_t, last_ctx)
+        if isinstance(result, Template):
+            result_t, middleware_api = result, None
+        elif isinstance(result, tuple):
+            if len(result) == 2:
+                if not isinstance(result[0], Template):
+                    raise TypeError(
+                        f"Component processor returned unxpected type in first entry of 2-tuple: {type(result[0])}"
+                    )
+                else:
+                    result_t, middleware_api = result
+            else:
+                raise ValueError(
+                    f"Component processor returned tuple with length != 2: {len(result)}"
+                )
+        else:
+            raise TypeError(
+                f"Component processor should return unexpected type: {type(result)}"
+            )
+
+        context_values: tuple[tuple[ContextVar, object], ...] = ()
+
+        if middleware_api is not None:
+            if isinstance(middleware_api, IMiddlewareGetContextValues):
+                context_values = middleware_api.get_context_values()
+            else:
+                # @DESIGN: It is NOT an error if a middleware object is provided but it
+                # provides no actual middleware functionality.  Should it be?
+                pass
+
+        # Try to consolidate "final" call(s) to the last block regardless of middleware.
+        if context_values:
+            with ContextVarSetter(context_values=context_values):
+                return self._process_template(result_t, last_ctx)
+        else:
+            return self._process_template(result_t, last_ctx)
 
     def _process_raw_texts(
         self,

tdom/processor_extension_test.py

@@ -101,7 +101,7 @@ def process(
             attrs: tuple[TAttribute, ...],
             component_template: Template,
             provided_attrs: tuple[Attribute, ...] = (),
-        ) -> Template:
+        ) -> Template | tuple[Template, object]:
             from inspect import isclass
 
             system_ctx = SystemCtx.get()
@@ -149,3 +149,96 @@ def test_replacement(self):
                 "</div>"
                 "</div>"
             )
+
+
+ThemeCtx: ContextVar[str] = ContextVar("ThemeCtx")
+
+ModeCtx: ContextVar[str] = ContextVar("ModeCtx")
+
+
+class TestMiddlewareGetContextValues:
+    @dataclass
+    class ThemeProvider:
+        theme_name: str
+
+        children: Template
+
+        mode: str | None = None
+
+        def __call__(self) -> tuple[Template, object]:
+            # Hit em' with another div...
+            result_t = t"<div>{self.children}</div>"
+            middleware_api = self
+            return result_t, middleware_api
+
+        def get_context_values(self) -> tuple[tuple[ContextVar, str], ...]:
+            context_values = ((ThemeCtx, self.theme_name),)
+            if self.mode is not None:
+                context_values += ((ModeCtx, self.mode),)
+            return context_values
+
+    @dataclass
+    class ThemeDisplay:
+        theme_name: str
+
+        mode: str | None = None
+
+        def __call__(self) -> Template:
+            sep = ":" if self.mode else None
+            return t"<span>{self.theme_name}{sep}{self.mode}</span>"
+
+    def _make_html(
+        self, default_theme_name: str = "theme-default", default_mode: str = "mode-dark"
+    ):
+
+        tp = TemplateProcessor()
+
+        def _html(template: Template, assume_ctx: ProcessContext | None = None):
+            if assume_ctx is None:
+                assume_ctx = ProcessContext()
+            with ThemeCtx.set(default_theme_name), ModeCtx.set(default_mode):
+                return tp.process(template, assume_ctx)
+
+        return _html
+
+    def _theme_name(self):
+        return ThemeCtx.get()
+
+    def _mode(self):
+        return ModeCtx.get()
+
+    def test_default(self):
+        html = self._make_html()
+        assert (
+            html(t"<{self.ThemeDisplay} theme_name={self._theme_name:callback} />")
+            == "<span>theme-default</span>"
+        )
+
+    def test_provider(self):
+        html = self._make_html()
+        child_t = t"<{self.ThemeDisplay} theme_name={self._theme_name:callback} />"
+        assert (
+            html(
+                t"<{self.ThemeProvider} theme_name='theme-pycon'>{child_t}</{self.ThemeProvider}>"
+            )
+            == "<div><span>theme-pycon</span></div>"
+        )
+
+    def test_provider_scope(self):
+        html = self._make_html()
+        child_t = t"<{self.ThemeDisplay} theme_name={self._theme_name:callback} />"
+        wrapped_t = t"<{self.ThemeProvider} theme_name='theme-pycon'>{child_t}</{self.ThemeProvider}>"
+        assert (
+            html(wrapped_t + child_t)
+            == "<div><span>theme-pycon</span></div><span>theme-default</span>"
+        )
+
+    def test_two_cvars(self):
+        html = self._make_html()
+        child_t = t"<{self.ThemeDisplay} theme_name={self._theme_name:callback} mode={self._mode:callback} />"
+        assert (
+            html(
+                t"<{self.ThemeProvider} theme_name='theme-pycon' mode='mode-light'>{child_t}</{self.ThemeProvider}>"
+            )
+            == "<div><span>theme-pycon:mode-light</span></div>"
+        )