Jerlshin's picture
defect fix career
f0894e2
Raw
History Blame Contribute Delete
2.33 kB
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)