protocols

triplum.cache.protocols

Semantic value, storage and serialization boundaries for computation caching.

Fingerprintable

Bases: Protocol

A semantic SHA-256 hex identity that distinguishes the declared value kind.

Select identity-bearing fields explicitly. Bookkeeping timestamps, cache resources and execution policies do not belong in this digest.

fingerprint abstractmethod

fingerprint() -> str

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

@abstractmethod
def fingerprint(self) -> str: ...

ComputationMetadata dataclass

ComputationMetadata(
    name: str,
    source_path: str | None = None,
    source_line: int | None = None,
)

Descriptive provenance, excluded from cache identity; location may be unavailable.

Attributes:

Name	Type	Description
name	str
source_path	str | None
source_line	int | None

name instance-attribute

name: str

source_path class-attribute instance-attribute

source_path: str | None = None

source_line class-attribute instance-attribute

source_line: int | None = None

from_callable classmethod

from_callable(compute: object) -> ComputationMetadata

Describe a callable without reading source files or invoking user code.

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

@classmethod
def from_callable(cls, compute: object) -> ComputationMetadata:
    """Describe a callable without reading source files or invoking user code."""
    target = compute.__func__ if isinstance(compute, MethodType) else compute
    if isinstance(target, (FunctionType, BuiltinFunctionType)):
        name = f"{target.__module__}.{target.__qualname__}"
    else:
        name = f"{type(target).__module__}.{type(target).__qualname__}"
    if isinstance(target, FunctionType) and not target.__code__.co_filename.startswith("<"):
        return cls(name, target.__code__.co_filename, target.__code__.co_firstlineno)
    return cls(name)

CacheKey dataclass

CacheKey(
    process: bytes,
    input: bytes,
    metadata: ComputationMetadata | None = None,
)

Full computation namespace and input data identity.

Attributes:

Name	Type	Description
process	bytes
input	bytes
metadata	ComputationMetadata | None

process instance-attribute

process: bytes

input instance-attribute

input: bytes

metadata class-attribute instance-attribute

metadata: ComputationMetadata | None = field(
    default=None, compare=False, hash=False
)

CachePolicy dataclass

CachePolicy(on_full: Literal['skip', 'block'] = 'skip')

Admission policy when accepted writes exhaust the pending-byte budget.

Attributes:

Name	Type	Description
on_full	Literal['skip', 'block']

on_full class-attribute instance-attribute

on_full: Literal['skip', 'block'] = 'skip'

Codec

Bases: Protocol

Serialize a value without changing its semantic content or fingerprint.

encode abstractmethod

encode(value: T) -> bytes

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

@abstractmethod
def encode(self, value: T, /) -> bytes: ...

decode abstractmethod

decode(payload: bytes) -> T

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

@abstractmethod
def decode(self, payload: bytes, /) -> T: ...

CacheBackend

Bases: Protocol

Storage supporting concurrent reads and one writer per Cache owner.

A completed put_many makes all its values visible to subsequent reads. Individual values are atomic; an error may leave a partial batch. close follows all calls.

get abstractmethod

get(key: CacheKey) -> bytes | None

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

@abstractmethod
def get(self, key: CacheKey, /) -> bytes | None: ...

put_many abstractmethod

put_many(entries: Sequence[tuple[CacheKey, bytes]]) -> None

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

@abstractmethod
def put_many(self, entries: Sequence[tuple[CacheKey, bytes]], /) -> None: ...

close abstractmethod

close() -> None

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

@abstractmethod
def close(self) -> None: ...