primus.utils.dependency_snapshots module

Dependency snapshot helpers for computed score consistency.

These helpers keep dependency consistency concerns outside Score.predict. They record exactly which child results a computed score used so audits can compare that snapshot with current persisted dependency results.

class primus.utils.dependency_snapshots.DependencySnapshot(dependencies: ~typing.List[~typing.Dict[str, ~typing.Any]], missing_dependencies: ~typing.List[~typing.Dict[str, ~typing.Any]] = <factory>, schema_version: str = 'dependency_snapshot_v1')

Bases: object

A normalized dependency snapshot for one computed score execution.

__init__(dependencies: ~typing.List[~typing.Dict[str, ~typing.Any]], missing_dependencies: ~typing.List[~typing.Dict[str, ~typing.Any]] = <factory>, schema_version: str = 'dependency_snapshot_v1') → None

as_dict() → Dict[str, Any]

property complete: bool

dependencies: List[Dict[str, Any]]

property hash: str

missing_dependencies: List[Dict[str, Any]]

schema_version: str = 'dependency_snapshot_v1'

async primus.utils.dependency_snapshots.check_score_result_dependency_snapshot_current(*, score_result: Any, client: Any, item_id: str, account_id: str | None) → Dict[str, Any]

Compare a cached computed result’s stored dependency snapshot to current persisted dependencies.

primus.utils.dependency_snapshots.dependency_configs_from_snapshot(snapshot: Dict[str, Any]) → List[Dict[str, Any]]

primus.utils.dependency_snapshots.dependency_entry_from_result(*, score_config: Dict[str, Any], result: Any, source: str, score_result_id: str | None = None, created_at: Any = None, updated_at: Any = None) → Dict[str, Any]

primus.utils.dependency_snapshots.dependency_snapshot_from_score_result(score_result: Any) → Dict[str, Any] | None

primus.utils.dependency_snapshots.excerpt(value: Any, limit: int = 500) → str | None

primus.utils.dependency_snapshots.extract_dependency_snapshot(value: Any) → Dict[str, Any] | None

async primus.utils.dependency_snapshots.fetch_persisted_dependency_snapshot(*, client: Any, item_id: str, account_id: str | None, dependency_configs: Sequence[Dict[str, Any]], result_type: str = 'prediction') → Tuple[DependencySnapshot, List[Score.Result]]

Fetch latest persisted ScoreResults for dependency configs.

primus.utils.dependency_snapshots.get_dependency_names(score_config: Dict[str, Any] | None) → List[str]

primus.utils.dependency_snapshots.get_score_config(scorecard_instance: Any, score_name: str) → Dict[str, Any] | None

primus.utils.dependency_snapshots.get_target_dependencies(scorecard_instance: Any, target_score_name: str) → List[Dict[str, Any]]

primus.utils.dependency_snapshots.isoformat(value: Any) → str | None

primus.utils.dependency_snapshots.json_object(value: Any) → Dict[str, Any]

primus.utils.dependency_snapshots.merge_snapshot_into_trace(trace: Dict[str, Any] | None, *, snapshot: DependencySnapshot | None) → Dict[str, Any] | None

primus.utils.dependency_snapshots.result_from_persisted_score_result(score_result: Any, score_config: Dict[str, Any]) → Score.Result

primus.utils.dependency_snapshots.score_version_id(score_config: Dict[str, Any] | None) → str | None

primus.utils.dependency_snapshots.snapshot_from_in_memory_results(*, scorecard_instance: Any, target_score_name: str, score_results: Dict[str, Any]) → DependencySnapshot

primus.utils.dependency_snapshots.stable_hash(value: Any) → str

primus.utils.dependency_snapshots.unwrap_result_entry(entry: Any) → Any

Normalize Scorecard result-map entries to their actual Score.Result value.