"""Connector for AWS S3.""" from urllib.parse import urlparse import boto3 from botocore.exceptions import ClientError from flask import g from payment.config import Config from payment.constants import constants def get_s3_client(): """Get S3 client.""" return boto3.client('s3') def create_presigned_url( bucket: str, client: boto3.client, key: str, expiration=constants.S3_PRESIGNED_URL_DEFAULT_EXPIRATION, ): """Generate a temporary presigned url to download a report from S3. Args: bucket (str): name of the s3 bucket client (boto3.client): instance of an s3 client key (str): name of existing s3 object key expiration (int): number of seconds the url is valid. Defaults to ten seconds. Returns: a presigned url (as a str) """ return client.generate_presigned_url( ClientMethod='get_object', ExpiresIn=expiration, Params={'Bucket': bucket, 'Key': key}, ) def get_object_metadata_by_s3_url(s3_url): """Return the metadata of the provided object on s3 bucket. Args: s3_url (str): full url of the object on s3 bucket Returns: Object's metadata (as a dict) """ bucket = Config.S3_PAYMENTS_BUCKET_NAME client = get_s3_client() parsed_url = urlparse(s3_url) key = parsed_url.path[1:] try: return client.head_object(Bucket=bucket, Key=key) except ClientError as err: g.log.error(err) return None