primus.utils.scoring module

class primus.utils.scoring.SingleScoreExecutionOutcome(result: Any | None, dependency_unmet: bool, score_results: Dict[str, Any], dependency_snapshot: Any | None = None, current_dependency_snapshot: Any | None = None, used_persisted_dependencies: bool = False)

Bases: object

Result envelope for single-score execution with dependency handling.

__init__(result: Any | None, dependency_unmet: bool, score_results: Dict[str, Any], dependency_snapshot: Any | None = None, current_dependency_snapshot: Any | None = None, used_persisted_dependencies: bool = False) → None

current_dependency_snapshot: Any | None = None

dependency_snapshot: Any | None = None

dependency_unmet: bool

result: Any | None

score_results: Dict[str, Any]

used_persisted_dependencies: bool = False

async primus.utils.scoring.check_if_score_is_disabled(scorecard_external_id: str, score_external_id: str, account_id: str) → bool

Check if a score is currently disabled by querying the API.

Args:

scorecard_external_id: The external ID of the scorecard score_external_id: The external ID of the score account_id: The DynamoDB ID of the account

Returns:

bool: True if the score is disabled, False otherwise

async primus.utils.scoring.create_score_result(item_id: str, scorecard_id: str, score_id: str, account_id: str, scoring_job_id: str, external_id: str, value: str, explanation: str, trace_data: dict = None, log_content: str = None, cost: dict = None, start_time_seconds: float | None = None, end_time_seconds: float | None = None, score_version_id: str | None = None, metadata_extra: dict | None = None, code: str = '200', status: str = 'COMPLETED', client: PrimusDashboardClient = None)

Create a ScoreResult in DynamoDB for caching worker results.

Args:

item_id: The DynamoDB Item ID scorecard_id: The DynamoDB Scorecard ID score_id: The DynamoDB Score ID account_id: The DynamoDB Account ID scoring_job_id: The DynamoDB ScoringJob ID external_id: The externalId of the Item value: The score value explanation: The explanation for the score trace_data: Optional trace data to upload to S3 log_content: Optional log content to upload to S3 cost: Optional cost data start_time_seconds: Optional evidence start timestamp in seconds end_time_seconds: Optional evidence end timestamp in seconds score_version_id: Optional score version ID to persist with the result metadata_extra: Additional metadata to merge into the ScoreResult metadata code: ScoreResult code field status: ScoreResult status field client: The PrimusDashboardClient instance

async primus.utils.scoring.create_scorecard_instance_for_single_score(scorecard_identifier: str, score_identifier: str) → Scorecard | None

Create a scorecard instance optimized for a single score request.

async primus.utils.scoring.enqueue_downstream_recompute_jobs(*, client: PrimusDashboardClient, account_id: str, item_id: str, scorecard_id: str, dependency_score_id: str, dependency_score_name: str, source_score_result_id: str | None = None) → list

Best-effort downstream recompute enqueue for scores depending on a fresh result.

async primus.utils.scoring.ensure_persisted_dependencies(*, scorecard_instance: Any, text: str, metadata: dict, modality: str, item: Any, client: Any, item_id: str, account_id: str, scorecard_id: str, scoring_job_id: str, external_id: str, dependency_configs: list, requested_score_id: str, requested_score_name: str, active_score_ids: set)

Compute and persist missing dependencies before a computed score runs.

async primus.utils.scoring.get_existing_score_result(report_id: str, scorecard_id: str, score_id: str, type: str, account_id: str) → dict | None

Check if a score result already exists for the given report, scorecard, and score. Now uses the new ScoreResult.find_by_cache_key method for cleaner, more maintainable code.

Args:

report_id: The external ID of the report scorecard_id: The external ID of the scorecard score_id: The external ID of the score account_id: The DynamoDB ID of the account

Returns:

A dictionary with the cached result if found, None otherwise

async primus.utils.scoring.get_external_id_from_item(item_id: str, client: PrimusDashboardClient) → str

Get externalId from an Item in DynamoDB by its item_id.

Args:

item_id: The DynamoDB ID of the Item client: The PrimusDashboardClient instance

Returns:

The externalId of the item, or None if not found

async primus.utils.scoring.get_metadata_from_item(item_id: str, client: PrimusDashboardClient) → dict

Get metadata from an Item in DynamoDB by its item_id.

Args:

item_id: The DynamoDB ID of the Item client: The PrimusDashboardClient instance

Returns:

The metadata dictionary, or empty dict if not found

async primus.utils.scoring.get_primus_client()

Get the Primus Dashboard client for API operations.

async primus.utils.scoring.get_text_from_item(item_id: str, client: PrimusDashboardClient) → str

Get text from an Item in DynamoDB by its item_id.

Args:

item_id: The DynamoDB ID of the Item client: The PrimusDashboardClient instance

Returns:

The text content of the item, or None if not found

async primus.utils.scoring.resolve_score_id(external_id: str, scorecard_dynamo_id: str, client) → Dict[str, str] | None

Resolve a score external ID to its DynamoDB ID and name, scoped by its parent scorecard’s DynamoDB ID.

Args:

external_id: The external ID of the score (e.g., “0”) scorecard_dynamo_id: The DynamoDB ID of the parent scorecard client: The dashboard client

Returns:

A dictionary with ‘id’ (DynamoDB ID) and ‘name’ of the score if found, None otherwise

async primus.utils.scoring.resolve_scorecard_id(external_id: str, account_id: str, client) → str | None

Resolve a scorecard external ID to its DynamoDB ID using the GSI, scoped by accountId.

Args:

external_id: The external ID of the scorecard account_id: The DynamoDB ID of the account client: The dashboard client

Returns:

The DynamoDB ID of the scorecard if found, None otherwise

primus.utils.scoring.sanitize_metadata_for_graphql(metadata: dict) → dict

Sanitize metadata for GraphQL compatibility.

Handles various data types to ensure the metadata can be safely stored in GraphQL/DynamoDB: - Removes problematic characters (null bytes, etc.) - Converts complex objects to JSON strings - Handles serialization errors gracefully

Note: Does NOT truncate data - all metadata is preserved in full.

Args:

metadata: Dictionary of metadata to sanitize

Returns:

Sanitized metadata dictionary safe for GraphQL operations

async primus.utils.scoring.score_single_target_with_dependencies(scorecard_instance, *, text: str, metadata: dict, modality: str, item: Any, target_score_id: str, target_score_name: str, client: Any = None, item_id: str | None = None, account_id: str | None = None, scorecard_id: str | None = None, scoring_job_id: str | None = None, external_id: str | None = None, requested_score_id: str | None = None, requested_score_name: str | None = None, _active_score_ids: set | None = None) → SingleScoreExecutionOutcome

Execute a single target score while preserving Scorecard dependency backfill behavior.

Returns a structured outcome so callers can distinguish dependency-unmet from hard failures.

async primus.utils.scoring.send_message_to_standard_scoring_request_queue(scoring_job_id: str)

Send message to AWS SQS queue with scoring_job_id