"""S3 common functions.""" import boto3 import botocore from src import config client = boto3.client('s3') resource = boto3.resource('s3') def head_object(bucket, key): """Retrieve object metadata without object downloading. Args: bucket (str): target S3 bucket. key (str): target object key. Returns: dict: object metadata. """ try: return client.head_object( Bucket=bucket, Key=key, ExpectedBucketOwner=config.EXPECTED_BUCKET_OWNER ) except botocore.exceptions.ClientError as e: if e.response['Error']['Code'] == '404': return None raise def download_file_object(bucket, key, file_object, **kwargs): """Download s3 object to file-like buffer. Args: bucket (str): target S3 bucket. key (str): target S3 key. file_object (io.BytesIO): file data. kwargs (dict): extra arguments. """ return client.download_fileobj( Bucket=bucket, Key=key, Fileobj=file_object, ExtraArgs={'ExpectedBucketOwner': config.EXPECTED_BUCKET_OWNER}, **kwargs )