Spaces:
Sleeping
Sleeping
File size: 2,325 Bytes
e1011bc | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 |
from __future__ import annotations
import re
from collections.abc import Mapping, Sequence
from datetime import date
from redstack.domain.enums import EvidenceKind
from redstack.domain.errors import ProvenanceError
from redstack.domain.provenance import EvidenceRef
#: The scalar union an EvidenceRef value may take (mirrors the canonical VO).
EvidenceValue = str | bool | int | float
_INDEX_RE = re.compile(r"^(?P<name>[^\[]+)\[(?P<idx>\d+)\]$")
def resolve_path(raw: Mapping[str, object], path: str) -> object:
cursor: object = raw
for segment in path.split("."):
match = _INDEX_RE.match(segment)
if match is not None:
name = match.group("name")
idx = int(match.group("idx"))
if not isinstance(cursor, Mapping) or name not in cursor:
raise ProvenanceError(f"path {path!r} segment {name!r} missing")
container = cursor[name]
if not isinstance(container, Sequence) or isinstance(container, (str, bytes)):
raise ProvenanceError(f"path {path!r} segment {name!r} not a list")
if idx >= len(container):
raise ProvenanceError(f"path {path!r} index {idx} out of range")
cursor = container[idx]
else:
if not isinstance(cursor, Mapping) or segment not in cursor:
raise ProvenanceError(f"path {path!r} segment {segment!r} missing")
cursor = cursor[segment]
return cursor
def mint(
raw: Mapping[str, object],
*,
kind: EvidenceKind,
path: str,
as_of: date | None = None,
) -> EvidenceRef:
"""Mint an EvidenceRef, verifying the path resolves to a JSON scalar.
Raises:
ProvenanceError: the path does not resolve, or resolves to a non-scalar.
"""
value = resolve_path(raw, path)
if isinstance(value, bool) or isinstance(value, (str, int, float)):
scalar: EvidenceValue = value
elif value is None:
# A null at a resolvable path is recorded as the literal string for audit
# (the path *exists*; its value is null), preserving the resolve guarantee.
scalar = "null"
else:
raise ProvenanceError(f"path {path!r} resolves to non-scalar {type(value).__name__}")
return EvidenceRef(kind=kind, path=path, value=scalar, as_of=as_of) |