| | |
- builtins.object
-
- S3FileClient
class S3FileClient(builtins.object) |
| |
S3FileClient(bucket_name: str, region: str = None, aws_access_key_id: str = None, aws_secret_access_key: str = None, error_collector: gen_ai_hub.evaluations.helpers.collector.ValidationCollector = None)
S3 client for read/write file operations with format-specific parsing.
|
| |
Methods defined here:
- __init__(self, bucket_name: str, region: str = None, aws_access_key_id: str = None, aws_secret_access_key: str = None, error_collector: gen_ai_hub.evaluations.helpers.collector.ValidationCollector = None)
- Initialize S3 client with flexible authentication options.
:param bucket_name: S3 bucket name
:type bucket_name: str
:param region: AWS region (defaults to boto3 default), defaults to None
:type region: str, optional
:param aws_access_key_id: AWS access key (optional if using IAM/profile), defaults to None
:type aws_access_key_id: str, optional
:param aws_secret_access_key: AWS secret key (optional if using IAM/profile), defaults to None
:type aws_secret_access_key: str, optional
:param error_collector: Validation error collector, defaults to None
:type error_collector: ValidationCollector, optional
- get_sqlitedb_tables_data_from_s3(self, s3_key: str, tables_list: List[str]) -> Dict[str, List[Dict]]
- Download SQLite DB from S3, load given tables into memory, return dict of lists.
:param s3_key: S3 object key for the SQLite database file
:type s3_key: str
:param tables_list: List of table names to extract from the database
:type tables_list: List[str]
:return: Dictionary mapping table names to lists of row dictionaries
:rtype: Dict[str, List[Dict]]
:raises RuntimeError: If database operations fail
- read_csv(self, s3_key: str, encoding: str = 'utf-8') -> List[Dict[str, Any]]
- Read CSV file from S3.
:param s3_key: S3 object key
:type s3_key: str
:param encoding: File encoding, defaults to "utf-8"
:type encoding: str, optional
:return: List of dictionaries, one per row
:rtype: List[Dict[str, Any]]
- read_json(self, s3_key: str, encoding: str = 'utf-8') -> Dict[str, Any]
- Read JSON file from S3.
:param s3_key: S3 object key
:type s3_key: str
:param encoding: File encoding, defaults to "utf-8"
:type encoding: str, optional
:return: Dictionary containing JSON data, or empty list if file is empty or error occurs
:rtype: Dict[str, Any]
- read_jsonl(self, s3_key: str, encoding: str = 'utf-8') -> List[Dict[str, Any]]
- Read JSONL (JSON Lines) file from S3.
:param s3_key: S3 object key
:type s3_key: str
:param encoding: File encoding, defaults to "utf-8"
:type encoding: str, optional
:return: List of dictionaries, one per line
:rtype: List[Dict[str, Any]]
- upload_csv(self, data: List[Dict], s3_key: str, **kwargs) -> bool
- Upload CSV data to S3.
:param data: List of dictionaries to upload as CSV
:type data: List[Dict]
:param s3_key: S3 key path
:type s3_key: str
:param kwargs: Additional S3 put_object parameters
:type kwargs: dict
:return: True if upload succeeded, False otherwise
:rtype: bool
- upload_json(self, data: Any, s3_key: str, **kwargs) -> bool
- Upload JSON data to S3.
:param data: Data to upload (will be JSON-serialized)
:type data: Any
:param s3_key: S3 key path
:type s3_key: str
:param kwargs: Additional S3 put_object parameters
:type kwargs: dict
:return: True if upload succeeded, False otherwise
:rtype: bool
- upload_jsonl(self, data: List[Dict], s3_key: str, **kwargs) -> bool
- Upload JSONL data to S3.
:param data: List of dictionaries to upload as JSONL
:type data: List[Dict]
:param s3_key: S3 key path
:type s3_key: str
:param kwargs: Additional S3 put_object parameters
:type kwargs: dict
:return: True if upload succeeded, False otherwise
:rtype: bool
Data descriptors defined here:
- __dict__
- dictionary for instance variables (if defined)
- __weakref__
- list of weak references to the object (if defined)
| |