plexus.utils.score_result_s3_utils module

plexus.utils.score_result_s3_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.score_result_s3_utils.download_score_result_log_file(s3_path, local_path=None)

Download a log file from S3.

Automatically detects the correct bucket based on the path: - Paths starting with ‘items/’ use the datasources bucket (item attachments) - Other paths use the score result attachments bucket

Args:

s3_path: S3 key path for the file (e.g., ‘items/123/transcript.txt’ or ‘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.score_result_s3_utils.download_score_result_trace_file(s3_path, local_path=None)

Download a trace file from S3.

Automatically detects the correct bucket based on the path: - Paths starting with ‘items/’ use the datasources bucket (item attachments) - Other paths use the score result attachments bucket

Args:

s3_path: S3 key path for the file (e.g., ‘items/123/deepgram.json’ or ‘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.score_result_s3_utils.get_bucket_name()

Get the S3 bucket name for score result attachments from environment variables or fall back to the default.

plexus.utils.score_result_s3_utils.get_datasources_bucket_name()

Get the S3 bucket name for data sources (item attachments) from environment variables or use default.

This bucket contains item attachments like Deepgram JSON files, transcripts, etc. The path format is typically: items/{item_external_id}/{filename}

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