utils

triplum.utils

Utilities independent of Triplum's domain and benchmark contracts.

FingerprintedComputationMixin

Identify a configured computation by loaded definitions and selected settings.

Implement fingerprint_config, returning {} for stateless computations. Referenced application helpers contribute automatically; select external content IDs as settings. No instance attributes are implicitly inspected. Override fingerprint for full control.

fingerprint_config

fingerprint_config() -> dict[str, object]

Select semantic settings; Paths identify names, not file contents.

Values support finite scalars, lists/tuples, string-keyed mappings, Paths and nested fingerprint() values. Mapping order is irrelevant: select ordered pairs explicitly if iteration order affects the computation.

Source code in src/triplum/utils/fingerprint.py

def fingerprint_config(self) -> dict[str, object]:
    """Select semantic settings; Paths identify names, not file contents.

    Values support finite scalars, lists/tuples, string-keyed mappings, Paths and
    nested fingerprint() values. Mapping order is irrelevant: select ordered pairs
    explicitly if iteration order affects the computation.
    """
    raise NotImplementedError("implement fingerprint_config() or override fingerprint()")

fingerprint

fingerprint() -> str

Return a SHA-256 digest; selected configuration is evaluated each time.

Source code in src/triplum/utils/fingerprint.py

def fingerprint(self) -> str:
    """Return a SHA-256 digest; selected configuration is evaluated each time."""
    config = self.fingerprint_config()
    if type(config) is not dict:
        raise TypeError("fingerprint_config() must return a string-keyed dictionary")
    return content_key(
        "configured-computation",
        {
            "definition": definition_hash(type(self)),
            "config": _state(config, _Definitions(type(self))),
        },
    )

canonical_json

canonical_json(obj: Any) -> str

Source code in src/triplum/utils/cache.py

def canonical_json(obj: Any) -> str:
    return json.dumps(obj, sort_keys=True, separators=(",", ":"), ensure_ascii=False)

content_key

content_key(kind: str, payload: Any) -> str

Source code in src/triplum/utils/cache.py

def content_key(kind: str, payload: Any) -> str:
    h = hashlib.sha256()
    h.update(kind.encode())
    h.update(b"\0")
    h.update(canonical_json(payload).encode())
    return h.hexdigest()

definition_hash cached

definition_hash(definition: object) -> str

Hash loaded Python methods, constants, defaults, captures and annotation definitions.

Includes inherited methods and domain Protocol bodies; excludes exact framework helpers. Application helpers and their referenced settings are followed statically. External libraries, mutable class attributes and schemas are not inferred. Definitions and settings must remain fixed after the first hash. Filename and source positions are ignored; interpreter/compiler upgrades can change the digest.

Source code in src/triplum/utils/fingerprint.py

@functools.cache
def definition_hash(definition: object) -> str:
    """Hash loaded Python methods, constants, defaults, captures and annotation definitions.

    Includes inherited methods and domain Protocol bodies; excludes exact framework helpers.
    Application helpers and their referenced settings are followed statically. External
    libraries, mutable class attributes and schemas are not inferred. Definitions and
    settings must remain fixed after the first hash. Filename and
    source positions are ignored; interpreter/compiler upgrades can change the digest.
    """
    return content_key("python-definition", _definition(definition, _Definitions(definition)))

source_hash cached

source_hash(cls: type) -> str

Hash of a class's source with comments, formatting and docstrings removed.

Source code in src/triplum/utils/fingerprint.py

@functools.cache
def source_hash(cls: type) -> str:
    """Hash of a class's source with comments, formatting and docstrings removed."""
    try:
        source = inspect.getsource(cls)
    except (OSError, TypeError) as error:
        raise TypeError(
            f"cannot fingerprint {cls.__qualname__}: its source is unavailable"
        ) from error
    tree = ast.parse(textwrap.dedent(source))
    for node in ast.walk(tree):
        if isinstance(node, (ast.ClassDef, ast.FunctionDef, ast.AsyncFunctionDef)):
            first = node.body[0]
            is_docstring = (
                isinstance(first, ast.Expr)
                and isinstance(first.value, ast.Constant)
                and isinstance(first.value.value, str)
            )
            if is_docstring:
                node.body = node.body[1:] or [ast.Pass()]
    return hashlib.sha256(ast.dump(tree).encode()).hexdigest()