"""S3 common functions.""" import boto3 import botocore from src import config client = boto3.client('s3') resource = boto3.resource('s3') def object_exists(bucket, key): """Check if S3 object exists. Args: bucket (str): target S3 bucket. key (str): target S3 key. """ try: 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 False else: raise else: return True 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, ExpectedBucketOwner=config.EXPECTED_BUCKET_OWNER) def create_presigned_url( bucket, key, client_method='get_object', expires_in=86400): """Generate a presigned url. Args: bucket (str): Bucket name. key (str): Existing S3 object key. client_method (str): Client method for operation by result URL. expires_in (int): The number of seconds the presigned url is valid for. By default it expires in 24 hours (86400 seconds). Returns: str: Presigned url. """ return client.generate_presigned_url( ClientMethod=client_method, ExpiresIn=expires_in, Params={ 'Bucket': bucket, 'Key': key } )