# tangl/core/template.py
# language=markdown
"""Templates (v38): compile/decompile and materialization.
This module defines the authoring boundary for core templates and separates
compile/decompile from runtime materialization.
See Also
--------
:class:`tangl.core.record.Record`
Template records inherit frozen/content/seq behavior from ``Record``.
:mod:`tangl.core.registry`
Registry-aware and hierarchical behaviors used by template containers.
Notes
-----
``NONGENERIC_FIELDS`` defines runtime-only fields stripped by ``decompile()``.
"""
from __future__ import annotations
from typing import Optional, Iterator, TypeVar, Generic, Type, Self, Any
from uuid import uuid4
import logging
from fnmatch import fnmatch
import re
from pydantic import Field
from tangl.type_hints import UnstructuredData, Identifier
from .entity import Entity
from .selector import Selector
from .registry import Registry, HierarchicalGroup, RegistryAware
from .record import Record
logger = logging.getLogger(__name__)
logger.setLevel(logging.WARNING)
ET = TypeVar("ET", bound=Entity)
NONGENERIC_FIELDS = {'uid', 'seq'} # discarded when decompiling to script
_BRACE_RE = re.compile(r"\{([^{}]+)\}")
MAX_SCOPE_BRACE_EXPANSIONS = 256
class _ScopeExpansionLimitError(ValueError):
"""Raised when admission-scope brace expansion exceeds configured limit."""
def _expand_scope_braces(
pattern: str,
*,
max_expansions: int = MAX_SCOPE_BRACE_EXPANSIONS,
) -> list[str]:
remaining = max_expansions
def _expand(value: str) -> list[str]:
nonlocal remaining
match = _BRACE_RE.search(value)
if match is None:
if remaining <= 0:
raise _ScopeExpansionLimitError(
"admission_scope brace expansion exceeds maximum allowed combinations"
)
remaining -= 1
return [value]
prefix = value[: match.start()]
suffix = value[match.end() :]
options = [opt.strip() for opt in match.group(1).split(",")]
expanded: list[str] = []
for option in options:
for tail in _expand(suffix):
expanded.append(f"{prefix}{option}{tail}")
return expanded
return _expand(pattern)
def _split_scope_path(path: str | None) -> list[str]:
if not isinstance(path, str) or not path:
return []
return [segment for segment in path.split(".") if segment]
def _scope_admitted_single(expanded_scope: str, ctx_parts: list[str]) -> bool:
"""Match scope prefix against a placement context with an implicit leaf.
``admission_scope`` is interpreted as a prefix over container/context segments.
The target context must include one additional trailing segment (the placement
leaf), so a scope like ``a.b`` admits ``a.b.c`` but not ``a.b``.
"""
scope_parts = _split_scope_path(expanded_scope)
if not scope_parts:
return True
if scope_parts[-1] in ("*", "**"):
prefix = scope_parts[:-1]
else:
prefix = scope_parts
if len(ctx_parts) <= len(prefix):
return False
for expected, actual in zip(prefix, ctx_parts):
if not fnmatch(actual, expected):
return False
return True
def _scope_admitted(template_scope: str | None, target_ctx: str | None) -> bool:
if template_scope in (None, "", "*"):
return True
if not isinstance(target_ctx, str) or not target_ctx:
return False
ctx_parts = _split_scope_path(target_ctx)
if not ctx_parts:
return False
try:
expanded_scopes = _expand_scope_braces(template_scope)
except _ScopeExpansionLimitError:
# Fail closed for pathological expansion inputs.
return False
for expanded in expanded_scopes:
if _scope_admitted_single(expanded, ctx_parts):
return True
return False
[docs]
class EntityTemplate(RegistryAware, Record, Generic[ET]):
"""EntityTemplate(payload: Entity)
Template wrapper around an entity payload.
Why
----
``EntityTemplate`` separates authoring-time prototypes from runtime
entities. A template can be compiled, searched, stored, and materialized
repeatedly without becoming part of the live graph itself.
Key Features
------------
* Supports authoring-loop transforms through ``compile`` and ``decompile``.
* Supports runtime instantiation through :meth:`materialize`.
* Distinguishes template-kind matching from payload-kind matching so callers
can query wrapper shape and produced entity kind independently.
* Can participate in template registries and hierarchical groups used by
provisioning and materialization.
API
---
- :meth:`has_template_kind` checks the wrapper record type.
- :meth:`has_payload_kind` checks the materialized entity kind.
- :meth:`has_kind` matches either axis as a convenience.
- :meth:`materialize` creates a live entity from the stored payload.
Example
-------
>>> class PseudoEntity(Entity): ...
>>> data = {'label': 'abc'}
>>> templ = EntityTemplate.from_data(data, default_kind=PseudoEntity)
>>> templ.has_template_kind(EntityTemplate) and templ.has_payload_kind(PseudoEntity)
True
>>> templ.materialize()
<PseudoEntity:abc>
>>> class PseudoEntity2(PseudoEntity): ...
>>> templ.materialize(kind=PseudoEntity2, label="def")
<PseudoEntity2:def>
>>> templ.materialize().uid != templ.payload.uid # fresh id by default
True
"""
# By default, templates are generic archetypes and may be used without restriction.
# In some cases, we want to impose other restrictions, for example, only allowing a
# template to be used within a scope, or once per scope. That should be captured in
# metadata.
payload: ET = Field(..., exclude=True)
# Excluded from pydantic model_dump; unstructure/structure handles payload explicitly.
admission_scope: str | None = None
# Optional target-context scope gate for provisioning admission.
def get_label(self):
"""Return template label, falling back to a payload-derived label."""
return self.label or f"from-{self.payload.get_label()}"
def get_hashable_content(self):
"""Use payload constructor-form data as template content identity."""
return {
"payload": self.payload.unstructure(),
"admission_scope": self.admission_scope,
}
@classmethod
def from_entity(cls, entity: Entity):
"""Build a template from a deep-evolved payload copy."""
return cls(payload=entity.evolve()) # holds a clean, deep copy
@classmethod
def from_data(cls, data: UnstructuredData, default_kind: Type[ET] = None) -> Self:
"""Build a template from constructor-form payload data."""
payload_data = dict(data)
if default_kind is not None:
payload_data.setdefault('kind', default_kind)
entity = Entity.structure(payload_data)
return cls.from_entity(entity)
# conflate/delegate identity matching
def has_kind(self, kind: Type[Entity]) -> bool:
"""Return ``True`` when kind matches template wrapper or payload kind."""
return super().has_kind(kind) or self.payload.has_kind(kind)
def has_template_kind(self, kind: Type[Entity]) -> bool:
"""Return ``True`` when kind matches only the template wrapper kind."""
return super().has_kind(kind)
def has_payload_kind(self, kind: Type[Entity]) -> bool:
"""Return ``True`` when kind matches only the payload kind."""
return self.payload.has_kind(kind)
def has_tags(self, *tags) -> bool:
"""Match tags against the union of template tags and payload tags."""
if len(tags) == 0:
return True
if len(tags) == 1 and tags[0] is None:
return True
if len(tags) == 1 and isinstance(tags[0], (tuple, list, set)):
tags = tuple(tags[0])
return set(tags).issubset(self.tags.union(self.payload.tags))
def get_identifiers(self) -> set[Identifier]:
"""Return combined identifier set from template and payload."""
return super().get_identifiers().union(self.payload.get_identifiers())
def admitted_to(self, target_ctx: str | None) -> bool:
"""Return whether this template admits provisioning at ``target_ctx``."""
return _scope_admitted(self.admission_scope, target_ctx)
# create copies
def materialize(self, preserve_uid: bool = False, **updates) -> ET:
"""Materialize a payload copy with optional overrides.
``kind`` overrides must narrow (subclass) relative to payload kind.
"""
# if preserve_uid is true
if 'kind' in updates:
if not issubclass(updates['kind'], self.payload.__class__):
raise TypeError(
"materialize kind must be a subclass of payload kind "
f"{self.payload.__class__.__name__}, got {updates['kind'].__name__}"
)
if not preserve_uid:
updates.setdefault('uid', uuid4()) # create a new uid if not provided
updates['templ_hash'] = self.content_hash() # indicate origin
else:
updates.pop('uid', None) # exact copy, discard any override uid
return self.payload.evolve(**updates)
def unstructure(self) -> UnstructuredData:
"""Serialize template record data plus explicitly serialized payload."""
data = super().unstructure()
# TODO: could use field annotation introspection to discover members and
# payload include nested entities and automatically structure/unstructure
# them recursively
data['payload'] = self.payload.unstructure()
return data
@classmethod
def structure(cls, data: UnstructuredData, _ctx=None) -> Self:
"""Structure template record data plus structured payload."""
data = dict(data)
data['payload'] = Entity.structure(data['payload'], _ctx=_ctx)
return super().structure(data)
def decompile(self, generify = True) -> UnstructuredData:
"""Return author-facing payload script data.
When ``generify`` is true, runtime-only fields in ``NONGENERIC_FIELDS`` are removed.
"""
# typically decompile will be used to go back to an author-facing
# script format, so we want to generify the payload as much as
# possible by removing irrelevant live instance fields.
data = self.payload.unstructure()
if generify:
for f in NONGENERIC_FIELDS:
data.pop(f, None)
if data.get('kind', None) is Entity:
# get rid of kind if it's redundant
# we will track this more extensively with 'explicit_fields'
# metadata during compile and defaults based on template subtypes
data.pop('kind', None)
return data
@classmethod
def compile(cls, data: UnstructuredData, _ctx=None) -> Self:
"""Compile author-facing payload script data into a template record."""
# Convenience for `structure(payload=<unstructured entity>)`
return cls.structure({'payload': data}, _ctx=_ctx)
[docs]
class TemplateRegistry(Registry[EntityTemplate]):
"""Registry of templates with convenience materialization and authoring helpers.
`TemplateRegistry` is the primary container for authoring-loop operations:
- `compile(script)` builds a flat registry from a list of script dicts.
- `decompile_all()` emits a list of script dicts from top-level template groups.
Materialization helpers (`materialize_one`, `materialize_all`) provide a simple bridge
into runtime, but they are not required for linting/compile/decompile.
Example:
>>> tr = TemplateRegistry()
>>> tr.add(EntityTemplate.from_data({'label': 'abc'}))
>>> tr.add(EntityTemplate.from_data({'label': 'def'}))
>>> tr.materialize_one(Selector.from_identifier('abc'))
<Entity:abc>
>>> list(tr.materialize_all())
[<Entity:abc>, <Entity:def>]
"""
def materialize_one(self, selector: Selector = None, sort_key=None, update: dict = None) -> Optional[ET]:
"""Materialize the first matching template, or ``None`` when not found."""
templ = self.find_one(selector=selector, sort_key=sort_key)
if templ is not None:
update = update or {}
return templ.materialize(**update)
def materialize_all(self, selector: Selector = None, sort_key=None) -> Iterator[ET]:
"""Materialize all matching templates lazily."""
# If you want to apply an update, do it one at a time.
templs = self.find_all(selector=selector, sort_key=sort_key)
return (templ.materialize() for templ in templs)
@classmethod
def compile(cls, data: list[UnstructuredData], _ctx=None, **kwargs) -> Self:
"""Compile top-level script list into a flat template registry."""
# this is 'registry = compile(script)' when script is a list of top-level template groups.
inst = cls(**kwargs)
for item in data:
item = dict(item)
factory = TemplateGroup if 'members' in item else EntityTemplate
template = factory.compile(data=item, _ctx=_ctx)
if isinstance(template, Iterator):
# Factory may yield multiple items.
for t in template:
inst.add(t)
else:
inst.add(template)
return inst
def decompile_all(self, generify = True) -> list[UnstructuredData]:
"""Decompile top-level template groups into author-facing script data."""
# this is 'script = decompile(registry)' when script is a list of top-level template groups
data: list[UnstructuredData] = []
top_level = self.find_all(Selector(
has_template_kind=TemplateGroup,
parent=None))
for item in top_level:
logger.debug(f"Decomposing tl item: {item!r} {item.parent!r}")
data.append(item.decompile(generify=generify))
return data
class TemplateGroup(EntityTemplate, HierarchicalGroup):
"""Template + hierarchical group membership for script-shaped trees.
A `TemplateGroup` is both:
- an `EntityTemplate` wrapping a payload (the group node), and
- a `HierarchicalGroup` whose membership is stored as `member_ids: list[UUID]`.
This enables *tree-shaped* scripts to be compiled into a flat registry and later
reconstructed.
## Representations
- **tree-IR**: author-facing dicts with inline `members`.
- **flat registry**: independent templates stored in a `TemplateRegistry`.
`TemplateGroup.compile()` performs the tree-IR → flat registry conversion:
- yields templates in **depth-first** order (children first)
- records **direct** children for each group via `member_ids`
`TemplateGroup.decompile()` performs the inverse projection:
- emits the group payload as a script dict
- recursively decompiles children into inline `members`
Note: multiple tree-IR shapes can map to the same flat registry unless additional
annotations are provided (e.g., kind-hints on member fields). v38 keeps the core
mechanism minimal; higher layers may add richer script parsing.
Example (tree-IR ⇄ flat registry round-trip):
>>> script = [
... { 'label': 'chapter-1',
... 'members': [
... { 'label': 'scene-1.1',
... 'members': [
... {'label': 'block-1.1.1'},
... {'label': 'block-1.1.2'} ]},
... { 'label': 'scene-1.2',
... 'members': [ {'label': 'block-1.2.1'} ] } ] } ]
>>> tr = TemplateRegistry.compile(script)
>>> len(tr)
6
>>> roundtrip = tr.decompile_all()
>>> assert script == roundtrip
"""
member_defaults: dict[str, Any] = Field(default_factory=dict)
# capture things like default-kind for members is "Scene" or scope is "parent-path.*"
# inject member defaults into members when structuring from payload, and exclude
# matching values from members when they are unstructured as payloads.
@classmethod
def compile(cls, data: UnstructuredData, _ctx=None) -> Iterator[EntityTemplate]:
"""Flatten a tree-ir payload into flat templates.
This yields templates in *depth-first* order (children first), while still
recording *direct* children for each group via `member_ids`.
Implementation trick: use a nested generator that `return`s the uid of the
direct root template for a payload subtree. Parent calls `child_id = yield from ...`.
"""
def _flatten(subtree: UnstructuredData) -> Iterator[EntityTemplate]:
subtree = dict(subtree)
members = subtree.pop('members', None)
# If this node has members, it is a TemplateGroup.
if members is not None:
member_ids: list[Any] = []
for child in members or []:
child_id = yield from _flatten(child)
member_ids.append(child_id)
# IMPORTANT: `member_ids` belong to the TemplateGroup record (HierarchicalGroup),
# not the payload. Payload stays "near-native".
group = cls.structure({'payload': subtree, 'member_ids': member_ids}, _ctx=_ctx)
yield group
return group.uid
# Otherwise this node is a plain EntityTemplate.
templ = EntityTemplate.structure({'payload': subtree}, _ctx=_ctx)
yield templ
return templ.uid
# Delegate to the nested generator.
yield from _flatten(data)
def decompile(self, generify: bool = True) -> UnstructuredData:
"""Decompile this group and recursively inline child member script entries."""
data = super().decompile(generify=generify)
data['members'] = []
for member in self.members():
data['members'].append(member.decompile(generify=generify))
return data
[docs]
class Snapshot(EntityTemplate):
"""Persistence convenience: a template that recreates an entity exactly.
A `Snapshot` is **not** part of the authoring loop. It is a persistence helper that
reuses the template/materialization machinery to recreate a live entity with the same
identifier and state.
- `materialize()` preserves uid and rejects updates.
- `decompile()` is not typically meaningful for snapshots.
Example:
>>> e = Entity(label='abc')
>>> s = Snapshot.from_entity(e)
>>> ee = s.materialize()
>>> e is not ee and e == ee # preserves uid
True
"""
def materialize(self, preserve_uid: bool = True, **updates) -> ET:
"""Materialize exact copy semantics; reject updates and uid replacement flags."""
if updates:
raise TypeError("Snapshot does not support updates")
if not preserve_uid:
raise TypeError("Snapshot does not support preserve_uid != True")
return super().materialize(preserve_uid=True)