Attachments
Attachments module.
This module provides attachment handling functionality for evaluation, including loading files from AWS S3 or local directories and managing temporary directories. It returns open binary file handles ready to be passed to inference functions or evaluators.
cleanup_temp_dirs(temp_dirs)
async
Cleanup temporary directories asynchronously.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
temp_dirs
|
list[TemporaryDirectory]
|
list[tempfile.TemporaryDirectory]: List of temporary directories to cleanup. |
required |
close_attachments_streams(streams)
Close all opened attachment streams if they are file handles.
This function is kept for backward compatibility. Since attachments are now stored as bytes, there are no streams to close. If BinaryIO handles are passed, they will be closed.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
streams
|
dict[str, BinaryIO] | dict[str, bytes] | None
|
dict[str, BinaryIO] | dict[str, bytes] | None: Mapping of file name to open binary file handle or bytes. If bytes, no action is taken. |
required |
download_gdrive_files(attachment_names, config)
async
Download multiple files from Google Drive concurrently.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
attachment_names
|
list[str]
|
List of file names to download |
required |
config
|
dict
|
Configuration with Google Drive settings |
required |
Returns:
| Type | Description |
|---|---|
tuple[list[tuple[str, str]], list[TemporaryDirectory]]
|
tuple[list[tuple[str, str]], list[tempfile.TemporaryDirectory]]: list of (original_name, local_file_path), temp_directories |
download_s3_files(attachment_names, config)
async
Download multiple files from S3 concurrently.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
attachment_names
|
list[str]
|
List of file names to download |
required |
config
|
dict
|
AWS configuration with S3 settings |
required |
Returns:
| Type | Description |
|---|---|
tuple[list[tuple[str, str]], list[TemporaryDirectory]]
|
tuple[list[tuple[str, str]], list[tempfile.TemporaryDirectory]]: list of (original_name, local_file_path), temp_directories |
load_attachments_streams(attachment_names, attachments_config)
async
Load attachments and return file bytes with the temp dirs.
Files are read into memory as bytes to avoid concurrency issues when multiple tasks access the same file handles. Bytes can be safely shared across concurrent tasks and wrapped in BytesIO when needed.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
attachment_names
|
Iterable[str]
|
Iterable of attachment file names to load |
required |
attachments_config
|
AttachmentConfig | dict[str, Any] | None
|
AttachmentConfig object specifying the type of attachment and the configuration for the attachment source. Must not be None if attachment_names are provided. |
required |
Returns:
| Type | Description |
|---|---|
tuple[dict[str, bytes], list[TemporaryDirectory]]
|
tuple[dict[str, bytes], list[tempfile.TemporaryDirectory]]: Tuple of (file_bytes_dict, temp_directories_list) |
Raises:
| Type | Description |
|---|---|
ValueError
|
If attachments_config is None/falsy when attachments are provided, or if the configuration is invalid |
open_local_files(attachment_names, base_dir=None)
Read local files into memory as bytes.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
attachment_names
|
Iterable[str]
|
Iterable of file names or relative paths. |
required |
base_dir
|
str | None
|
Optional base directory to prepend to each name. |
None
|
Returns:
| Type | Description |
|---|---|
dict[str, bytes]
|
dict[str, bytes]: Mapping of file name to file bytes. |