plexus.utils.scoring module

async plexus.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 plexus.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, client: PlexusDashboardClient = 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 client: The PlexusDashboardClient instance

async plexus.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 plexus.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 plexus.utils.scoring.get_external_id_from_item(item_id: str, client: PlexusDashboardClient) → str

Get externalId from an Item in DynamoDB by its item_id.

Args:

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

Returns:

The externalId of the item, or None if not found

async plexus.utils.scoring.get_metadata_from_item(item_id: str, client: PlexusDashboardClient) → dict

Get metadata from an Item in DynamoDB by its item_id.

Args:

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

Returns:

The metadata dictionary, or empty dict if not found

async plexus.utils.scoring.get_plexus_client()

Get the Plexus Dashboard client for API operations.

async plexus.utils.scoring.get_text_from_item(item_id: str, client: PlexusDashboardClient) → str

Get text from an Item in DynamoDB by its item_id.

Args:

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

Returns:

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

async plexus.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 plexus.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

plexus.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 plexus.utils.scoring.send_message_to_standard_scoring_request_queue(scoring_job_id: str)

Send message to AWS SQS queue with scoring_job_id