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