Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
107 changes: 78 additions & 29 deletions src/deigma/template.py
Original file line number Diff line number Diff line change
Expand Up @@ -67,7 +67,12 @@ def template(
source = load_template_source(path)

if source is not None:
return inline_template(source, serialize=serialize, use_proxy=use_proxy)
return inline_template(
source,
serialize=serialize,
use_proxy=use_proxy,
_path=str(path) if path else None,
)

raise ValueError("Invalid arguments")

Expand All @@ -87,33 +92,33 @@ def inline_template(
*,
serialize: Serialize = DEFAULT_SERIALIZE,
use_proxy: bool = USE_PROXY,
_path: str | None = None,
) -> Callable[[type[T]], type[Template]]:
def decorator(cls: type[T]) -> type[T]:
config = ConfigDict(arbitrary_types_allowed=True)
cls = pydantic_dataclass(init=False, config=config)(
type(cls.__name__, (cls,), dict(cls.__dict__))
)
cls.__is_deigma_template__ = True
cls._source = cleandoc(source)

_src = cleandoc(source)
engine = Jinja2Engine(serialize=serialize)
cls._engine = engine
cls._variables = engine.introspect_variables(source)
variables = tuple(engine.introspect_variables(_src))

static_fields = set(cls.__annotations__)
properties = {
prop for prop in vars(cls) if isinstance(getattr(cls, prop), property)
}
fields = static_fields | properties

if not set(cls._variables).issubset(fields):
variables = cls._variables
if not set(variables).issubset(fields):
msg = (
"Template variables mismatch. Template fields must match variables in source:\n\n"
f"fields on type: {fields}, variables in source: {variables}"
)

if any(source.endswith(ext) for ext in COMMON_JINJA2_EXTENSIONS):
extension = source.split(".")[-1]
if any(_src.endswith(ext) for ext in COMMON_JINJA2_EXTENSIONS):
extension = _src.split(".")[-1]
msg += (
f"\n\nHint: 'source' ends with '.{extension}', "
"which is a common Jinja2 extension. "
Expand All @@ -123,39 +128,83 @@ def decorator(cls: type[T]) -> type[T]:

raise ValueError(msg)

cls._compiled_template = engine.compile_template(cls._source)
cls._type_adapter = TypeAdapter(cls)
def _get_adapter(c):
ta = getattr(c, "_type_adapter", None)
if ta is None:
ta = TypeAdapter(c)
setattr(c, "_type_adapter", ta)
return ta

def _get_compiled_template(cache):
# Optional file mtime check for hot-reload
if cache["path"] and os.path.exists(cache["path"]):
mtime = os.path.getmtime(cache["path"])
if cache["mtime"] != mtime:
cache["src"] = load_template_source(cache["path"])
cache["tmpl"] = None # force recompile
cache["mtime"] = mtime

if cache["tmpl"] is None:
cache["tmpl"] = cache["engine"].compile_template(cache["src"])
return cache["tmpl"]

if use_proxy:

def __str__(instance):
proxied = {
field: getattr(instance._proxy, field) for field in cls._variables
def __str__(
instance,
_cache={
"tmpl": None,
"engine": Jinja2Engine(serialize=serialize),
"src": _src,
"vars": variables,
"path": _path,
"mtime": None,
},
):
tmpl = _get_compiled_template(_cache)

if not hasattr(instance, "_proxy"):
adapter = _get_adapter(instance.__class__)
instance._proxy = SerializationProxy.build(instance, adapter)

proxied = {name: getattr(instance._proxy, name) for name in _cache["vars"]}
return tmpl.render(proxied)

else:

def __str__(
instance,
_cache={
"tmpl": None,
"engine": Jinja2Engine(serialize=serialize),
"src": _src,
"vars": variables,
"path": _path,
"mtime": None,
},
):
tmpl = _get_compiled_template(_cache)

adapter = _get_adapter(instance.__class__)
serialized = adapter.dump_python(instance)
rendered_fields = {
name: _render_field_maybe(getattr(instance, name), serialized[name])
for name in _cache["vars"]
}
return instance._compiled_template.render(proxied)
return tmpl.render(rendered_fields)

cls.__str__ = __str__

if use_proxy:
original_init = cls.__init__

def __init__(instance, *args, **kwargs):
original_init(instance, *args, **kwargs)
instance._proxy = SerializationProxy.build(instance, cls._type_adapter)
adapter = _get_adapter(cls)
instance._proxy = SerializationProxy.build(instance, adapter)

cls.__init__ = __init__

else:

def __str__(instance):
serialized = cls._type_adapter.dump_python(instance)
rendered_fields = {
field: _render_field_maybe(
getattr(instance, field), serialized[field]
)
for field in cls._variables
}
return instance._compiled_template.render(rendered_fields)

cls.__str__ = __str__

return cls

return decorator
Expand Down
238 changes: 238 additions & 0 deletions tests/benches/test_template_overhead.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,238 @@
"""
Benchmarks for template compilation and rendering overhead.

These benchmarks measure the overhead introduced by the method-defaults
approach with lazy compilation and hot-reload features.
"""

from pathlib import Path

import pytest

from deigma import template


# Fixtures
@pytest.fixture
def simple_inline_template():
"""Simple inline template for benchmarks."""

@template("Hello, {{ name }}!")
class HelloTemplate:
name: str

return HelloTemplate


@pytest.fixture
def file_template(tmp_path: Path):
"""File-backed template for benchmarks."""
template_file = tmp_path / "test.jinja2"
template_file.write_text("Hello, {{ name }}!")

@template(path=template_file)
class HelloTemplate:
name: str

return HelloTemplate


@pytest.fixture
def complex_inline_template():
"""More complex template for realistic benchmarks."""

@template(
"""
{% for item in items %}
- {{ item.name }}: {{ item.value }}
{% endfor %}
"""
)
class ItemListTemplate:
items: list[dict]

return ItemListTemplate


# Template Creation Benchmarks
def test_benchmark_template_definition(benchmark):
"""Benchmark template class definition (happens once)."""

def define_template():
@template("Hello, {{ name }}!")
class HelloTemplate:
name: str

return HelloTemplate

benchmark(define_template)


def test_benchmark_instance_creation(benchmark, simple_inline_template):
"""Benchmark creating template instance."""
benchmark(simple_inline_template, name="World")


# First Render (Lazy Compilation) Benchmarks
def test_benchmark_first_render_inline(benchmark, simple_inline_template):
"""Benchmark first render of inline template (triggers lazy compilation)."""

def first_render():
instance = simple_inline_template(name="World")
return str(instance)

benchmark(first_render)


def test_benchmark_first_render_file(benchmark, file_template):
"""Benchmark first render of file template (includes mtime check + lazy compilation)."""

def first_render():
instance = file_template(name="World")
return str(instance)

benchmark(first_render)


# Subsequent Renders (Cached) Benchmarks
def test_benchmark_cached_render_inline(benchmark, simple_inline_template):
"""Benchmark subsequent renders of inline template (uses cached compiled template)."""
instance = simple_inline_template(name="World")
# First render to populate cache
str(instance)

# Benchmark cached renders
benchmark(str, instance)


def test_benchmark_cached_render_file(benchmark, file_template):
"""Benchmark subsequent renders of file template (cached + mtime check)."""
instance = file_template(name="World")
# First render to populate cache
str(instance)

# Benchmark cached renders (includes mtime check)
benchmark(str, instance)


# Comparison: Multiple Instances
def test_benchmark_multiple_instances_inline(benchmark, simple_inline_template):
"""Benchmark rendering multiple instances of same template (share cache)."""

def render_multiple():
results = []
for i in range(10):
instance = simple_inline_template(name=f"User{i}")
results.append(str(instance))
return results

benchmark(render_multiple)


def test_benchmark_multiple_instances_file(benchmark, file_template):
"""Benchmark rendering multiple instances of file template (share cache + mtime checks)."""

def render_multiple():
results = []
for i in range(10):
instance = file_template(name=f"User{i}")
results.append(str(instance))
return results

benchmark(render_multiple)


# Complex Template Benchmarks
def test_benchmark_complex_template_first_render(benchmark, complex_inline_template):
"""Benchmark first render of complex template."""

def first_render():
items = [{"name": f"item{i}", "value": i} for i in range(20)]
instance = complex_inline_template(items=items)
return str(instance)

benchmark(first_render)


def test_benchmark_complex_template_cached_render(benchmark, complex_inline_template):
"""Benchmark cached render of complex template."""
items = [{"name": f"item{i}", "value": i} for i in range(20)]
instance = complex_inline_template(items=items)
# First render to populate cache
str(instance)

# Benchmark cached renders
benchmark(str, instance)


# Overhead Measurement: Same Instance Multiple Renders
def test_benchmark_same_instance_repeated_renders_inline(
benchmark, simple_inline_template
):
"""Benchmark rendering same instance multiple times (inline template)."""
instance = simple_inline_template(name="World")

def repeated_renders():
for _ in range(100):
str(instance)

benchmark(repeated_renders)


def test_benchmark_same_instance_repeated_renders_file(benchmark, file_template):
"""Benchmark rendering same instance multiple times (file template with mtime checks)."""
instance = file_template(name="World")

def repeated_renders():
for _ in range(100):
str(instance)

benchmark(repeated_renders)


# Mtime Check Overhead
def test_benchmark_mtime_check_overhead(benchmark, tmp_path: Path):
"""Benchmark the overhead of mtime checking in file templates."""
# Create a file template
template_file = tmp_path / "bench.jinja2"
template_file.write_text("Value: {{ value }}")

@template(path=template_file)
class BenchTemplate:
value: int

# Warm up cache
instance = BenchTemplate(value=42)
str(instance)

# Benchmark renders with mtime check
def render_with_mtime_check():
return str(instance)

benchmark(render_with_mtime_check)


# Lazy Proxy Building Overhead
def test_benchmark_lazy_proxy_building(benchmark):
"""Benchmark lazy proxy building for old instances."""

@template("{{ value }}")
class ValueTemplate:
value: int

# Create instance
instance = ValueTemplate(value=42)
# Don't render yet

# Simulate old instance scenario: delete _proxy if it exists
# This tests the lazy proxy building path
if hasattr(instance, "_proxy"):
delattr(instance, "_proxy")

# Benchmark first render which builds proxy
def render_building_proxy():
if hasattr(instance, "_proxy"):
delattr(instance, "_proxy")
return str(instance)

benchmark(render_building_proxy)
Loading