"""S3 common functions.""" import boto3 from botocore import exceptions from oto import response from constants import errors client = boto3.client('s3') resource = boto3.resource('s3') def get_object(bucket, key): """Get S3 object. Args: bucket (str): target S3 bucket. key (str): target S3 key. """ return client.get_object(Bucket=bucket, Key=key) def object_exists(bucket, key): """Check if S3 object exists. Args: bucket (str): target S3 bucket. key (str): target S3 key. Return: response.Response: Response object with success or error details. """ try: client.head_object(Bucket=bucket, Key=key) return response.Response({'status': errors.SUCCESS_CODE}) except exceptions.ClientError as e: error_code = e.response['Error']['Code'] if error_code == '404': error_message = errors.ERROR_MESSAGE_S3_FILE_NOT_FOUND.format( key=key, bucket=bucket) return response.create_not_found_response(error_message) elif error_code == '403': error_message = errors.ERROR_MESSAGE_S3_FILE_ACCESS_DENIED.format( key=key, bucket=bucket) return response.create_error_response( errors.ERROR_CODE_S3_FILE_ACCESS_DENIED, error_message, status=403) else: error_message = errors.ERROR_MESSAGE_S3_FILE_FAILED_TO_HEAD.format( key=key, bucket=bucket, error=str(e)) return response.create_fatal_response(error_message) def head_object(bucket, key): """Retrieve object metadata without object downloading. Args: bucket (str): target S3 bucket. key (str): target object key. Returns: dict: Response from S3. """ return client.head_object(Bucket=bucket, Key=key) def download_object(bucket, key, local_filename): """Download S3 object from a bucket. Args: bucket (str): target S3 bucket. key (str): target object key. local_filename (str): path where file should be saved. Returns: str: path to downloaded file. """ my_bucket = resource.Bucket(bucket) my_bucket.download_file(key, local_filename) return local_filename