plexus.utils package

plexus.utils.batch_find_items_by_identifiers(identifier_values: List[str], account_id: str, client: PlexusDashboardClient) → Dict[str, Item]

Efficiently find multiple items by their identifier values.

Uses batch lookup for optimal performance when searching for many items.

Args:

identifier_values: List of exact values to search for account_id: Account to search within client: API client

Returns:

Dictionary mapping identifier_value -> Item for found items. Missing identifier values will not appear in the result.

Example:
items_dict = batch_find_items_by_identifiers(

[“CUST-123456”, “ORD-789012”, “TICKET-555”], “account123”, client

)

for value, item in items_dict.items():

print(f”Identifier {value} -> Item {item.id}”)

# Check if specific identifier was found if “CUST-123456” in items_dict:

customer_item = items_dict[“CUST-123456”]

plexus.utils.capture_request_logs(request_id: str | None = None)

Context manager to capture logs for a specific request.

Usage:
with capture_request_logs() as (request_id, get_logs):

# Do work that generates logs logs = get_logs()

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

plexus.utils.check_s3_bucket_access(bucket_name=None)

Diagnostic function to check if the S3 bucket exists and is accessible.

Args:

bucket_name: Name of the bucket to check, or None to use the default

Returns:

Dict with diagnostic information

plexus.utils.create_identifiers_for_item(item_id: str, account_id: str, identifiers_data: List[Dict[str, Any]], client: PlexusDashboardClient) → List[Identifier]

Create identifier records for an item from JSON-style data.

Converts from the old JSON format to separate Identifier records. Useful for migration and dual-write scenarios.

Args:

item_id: ID of the item these identifiers belong to account_id: Account ID for data isolation identifiers_data: List of identifier dicts in JSON format:

[{“name”: “Customer ID”, “id”: “CUST-123”, “url”: “…”}]

client: API client

Returns:

List of created Identifier objects (may be empty if all fail)

Example:

# Convert from JSON format to Identifier records json_identifiers = [

{“name”: “Customer ID”, “id”: “CUST-123456”, “url”: “https://…”}, {“name”: “Order ID”, “id”: “ORD-789012”}

]

identifier_records = create_identifiers_for_item(

“item123”, “account123”, json_identifiers, client

)

async plexus.utils.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.create_scorecard_instance_for_single_score(scorecard_identifier: str, score_identifier: str) → Scorecard | None

Create a scorecard instance optimized for a single score request.

plexus.utils.download_score_result_log_file(s3_path, local_path=None)

Download a log file from the score result attachments S3 bucket.

Args:

s3_path: S3 key path for the file (e.g., ‘scoreresults/123/log.txt’) local_path: Optional local path to save the file to

Returns:

The content of the file as a string, and the local path if saved

plexus.utils.download_score_result_trace_file(s3_path, local_path=None)

Download a trace file from the score result attachments S3 bucket.

Args:

s3_path: S3 key path for the file (e.g., ‘scoreresults/123/trace.json’) local_path: Optional local path to save the file to

Returns:

The content of the file as a dictionary (parsed JSON), and the local path if saved

plexus.utils.find_item_by_identifier(identifier_value: str, account_id: str, client: PlexusDashboardClient) → Item | None

Find an item by exact identifier value within an account.

This is the primary search function for exact-match lookups. Uses O(1) GSI lookup for consistent performance.

Args:

identifier_value: Exact value to search for (e.g., “CUST-123456”) account_id: Account to search within client: API client

Returns:

Item object if found, None if not found

Example:

item = find_item_by_identifier(“CUST-123456”, “account123”, client) if item:

print(f”Found item {item.id} with data: {item.data}”)

plexus.utils.find_item_by_typed_identifier(identifier_name: str, identifier_value: str, account_id: str, client: PlexusDashboardClient) → Item | None

Find an item by identifier name and exact value.

More specific than find_item_by_identifier when you know the identifier type. Useful when the same value might exist for different identifier types.

Args:

identifier_name: Name of identifier type (e.g., “Customer ID”) identifier_value: Exact value to search for (e.g., “123456”) account_id: Account to search within client: API client

Returns:

Item object if found, None if not found

Example:

# Find item with specifically a “Customer ID” of “123456” # (ignoring any “Order ID” or other types with same value) item = find_item_by_typed_identifier(

“Customer ID”, “123456”, “account123”, client

)

plexus.utils.find_items_by_identifier_type(identifier_name: str, account_id: str, client: PlexusDashboardClient, limit: int | None = None) → List[Item]

Find all items that have identifiers of a specific type.

Args:

identifier_name: Name of identifier type (e.g., “Customer ID”, “Order ID”) account_id: Account to search within client: API client limit: Optional limit on number of results

Returns:

List of Item objects that have identifiers of the specified type

Example:

# Find all items with “Customer ID” identifiers items = find_items_by_identifier_type(“Customer ID”, “account123”, client) for item in items:

print(f”Item {item.id} has customer identifier”)

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

plexus.utils.get_item_identifiers(item_id: str, client: PlexusDashboardClient) → List[Dict[str, Any]]

Get all identifiers for a specific item in a format compatible with the UI.

Returns identifiers in the same format as the deprecated JSON field, ensuring compatibility with existing UI components.

Args:

item_id: ID of the item to get identifiers for client: API client

Returns:

List of identifier dictionaries with ‘name’, ‘id’ (value), and optional ‘url’

Example:

identifiers = get_item_identifiers(“item123”, client) # Returns: [ # {“name”: “Customer ID”, “id”: “CUST-123456”, “url”: “https://…”}, # {“name”: “Order ID”, “id”: “ORD-789012”} # ]

async plexus.utils.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.get_plexus_client()

Get the Plexus Dashboard client for API operations.

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

Send message to AWS SQS queue with scoring_job_id

plexus.utils.truncate_dict_strings(data: Dict[str, Any] | List[Dict[str, Any]], max_length: int = 100, truncation_indicator: str = '...') → Dict[str, Any] | List[Dict[str, Any]]
plexus.utils.truncate_dict_strings_inner(data: Dict[str, Any], max_length: int = 100, truncation_indicator: str = '...') → Dict[str, Any]
plexus.utils.upload_score_result_log_file(score_result_id: str, log_content: str) → str | None

Upload a log file to S3 for a ScoreResult.

Args:

score_result_id: The ScoreResult ID log_content: The log content to upload

Returns:

The S3 path (key) if successful, None otherwise

plexus.utils.upload_score_result_trace_file(score_result_id, trace_data, file_name='trace.json')

Upload trace data as a JSON file to the score result attachments S3 bucket.

Args:

score_result_id: ID of the score result this file belongs to trace_data: Dictionary or JSON-serializable data to upload file_name: Name of the file to create (default: “trace.json”)

Returns:

The S3 path (key) for the file, formatted for Amplify Gen2

Submodules