numeraire_dataset.zones.DataLock#

class numeraire_dataset.zones.DataLock(artifacts: dict[str, dict[str, ~typing.Any]]=<factory>)[source]#

An in-memory data.lock.json with typed add/read helpers.

__init__(artifacts: dict[str, dict[str, ~typing.Any]]=<factory>) → None#

Methods

`__init__`(artifacts, dict[str, ~typing.Any]] =)
`add_clean`(built, *, inputs[, step_version])	Record an auditable clean recipe with role-bound artifact labels and hashes.
`add_raw`(source, vintage, content_digest, *)	Record a pinned raw vintage; returns its artifact key `<source>@<vintage>`.
`data_vintage`(name)	The `data_vintage` string of a recorded clean artifact (the numeraire-facing stamp).
`read`(path)	Load a lock file (accepts a directory containing `data.lock.json`).
`to_json`()	Serialize to canonical (sorted-key) JSON.
`write`(path)	Write `data.lock.json` (or the given path) and return it.

Attributes

artifacts

add_raw(source: str, vintage: str, content_digest: str, *, query_hash: str = '', pit_status: str = 'vintage', content_digest_version: int = 2) → str[source]#: Record a pinned raw vintage; returns its artifact key <source>@<vintage>.

add_clean(built: Built, *, inputs: Mapping[str, str], step_version: int | None = None) → str[source]#: Record an auditable clean recipe with role-bound artifact labels and hashes.

data_vintage(name: str) → str[source]#: The data_vintage string of a recorded clean artifact (the numeraire-facing stamp).

write(path: str | Path) → Path[source]#: Write data.lock.json (or the given path) and return it.

classmethod read(path: str | Path) → DataLock[source]#: Load a lock file (accepts a directory containing data.lock.json).