"""S3 common functions.""" import boto3 import botocore client = boto3.client('s3') resource = boto3.resource('s3') def upload_key(bucket, key, data): """Upload S3 key. Args: bucket (str): target S3 bucket. key (str): target S3 key. data (bytes): key content. """ client.put_object(Bucket=bucket, Key=key, Body=data) def copy_key(source_bucket, source_key, new_bucket, new_key): """Copy S3 key. Args: source_bucket (str): source S3 bucket. source_key (str): source S3 key. source_bucket (str): new S3 bucket. source_key (str): new S3 key. """ bucket = resource.Bucket(new_bucket) bucket.copy({'Bucket': source_bucket, 'Key': source_key}, new_key) def delete_key(bucket, key): """Delete S3 key. Args: bucket (str): target S3 bucket. key (str): target S3 key. """ client.delete_object(Bucket=bucket, Key=key) def key_exists(bucket, key): """Check if S3 key exists. Args: bucket (str): target S3 bucket. key (str): target S3 key. """ try: client.head_object(Bucket=bucket, Key=key) except botocore.exceptions.ClientError as e: if e.response['Error']['Code'] == '404': return False else: raise else: return True