# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. # SPDX-License-Identifier: Apache-2.0 """ Helper functions for S3 """ import boto3 from botocore.client import Config from typing import List def get_s3_client(): """ Get a default S3 client, configured to use Virtual Hosted-Style addressing. Virtual Hosted-Style addressing ensures pre-signed URLs generated by this client will avoid potential HTTP 307 Temporary Redirect errors that may occur within the first 24 hours after the bucket is first created. See: https://aws.amazon.com/premiumsupport/knowledge-center/s3-http-307-response/ DO NOT USE this client if your bucket is configured with CORS (Cross-Origin Resource Sharing), otherwise you may face errors during the first several hours after the bucket is created due to DNS changes having not propagated yet. For CORS-configured buckets, you need to initially use Path-Style addressing. See: https://boto3.amazonaws.com/v1/documentation/api/1.9.42/guide/s3.html#changing-the-addressing-style For more information on S3 addressing styles, please see: https://docs.aws.amazon.com/AmazonS3/latest/userguide/VirtualHosting.html """ return boto3.client( service_name='s3', config=Config(s3={'addressing_style': 'virtual'}) ) def get_bucket(bucket_name: str): return boto3.resource('s3', config=Config(s3={'addressing_style': 'path'})).Bucket(bucket_name) def delete_items(keys: List[str], s3_bucket): """Delete the specified keys from the provided S3 bucket and return the S3 response.""" delete_request = { 'Objects': [{'Key': key} for key in keys] } print(f'Deleting keys {keys} from bucket {s3_bucket.name}') return s3_bucket.delete_objects(Delete=delete_request)