"""Connector to S3 buckets.""" import logging from typing import cast import boto3 from botocore.exceptions import ClientError logger = logging.getLogger(__name__) class FailedDownloadException(Exception): """Raised when file download fails.""" pass class S3: """Connect to S3 bucket and read files.""" def __init__(self) -> None: """Initialize the S3 client using provided AWS credentials.""" self._client = boto3.client("s3") def does_file_exist(self, bucket: str, key: str) -> bool: """Check if a file exists in the S3 bucket using head_object.""" try: response = self._client.head_object(Bucket=bucket, Key=key) return "ResponseMetadata" in response except ClientError as e: if e.response["Error"]["Code"] == "404": return False logger.warning( "Failed to check existence of file in bucket %s with key %s", bucket, key, ) raise e def download_file(self, bucket: str, key: str, local_path: str) -> str: """Download a file from S3 bucket to a local path.""" try: self._client.download_file(bucket, key, local_path) return local_path except ClientError as e: logger.warning( "Failed to download file from bucket %s with key %s", bucket, key ) raise FailedDownloadException( "Failed to download file from bucket %s with key %s", bucket, key ) from e def get_file_content(self, bucket: str, key: str) -> str: """Read passed file from a bucket.""" try: response = self._client.get_object(Bucket=bucket, Key=key) body = cast(bytes, response["Body"].read()) return body.decode(encoding="utf-8-sig") except ClientError as e: logger.warning( "Failed to read file content from bucket %s with key %s", bucket, key ) raise e