"""Connector for AWS S3 service.""" import boto3 from botocore.exceptions import ClientError, NoCredentialsError from oto import response import sentry_sdk from project_manager import config from project_manager.constant import error_const def get_s3_file_url(filename): """Generate tokenized s3 url for provided filename. Args: filename(str): name of file which should have tokenized url. Returns: response.Response: response object with url in message. """ try: client = boto3.client('s3') bucket_name, file_path = config.S3_LCE_BUCKET_PATH.split( '/', maxsplit=1) key = '{file_path}/{filename}'.format( file_path=file_path, filename=filename) client.head_bucket(Bucket=bucket_name) client.head_object(Bucket=bucket_name, Key=key) url = client.generate_presigned_url( 'get_object', Params={ 'Bucket': bucket_name, 'Key': key}, ExpiresIn=86400) except NoCredentialsError: sentry_sdk.capture_exception() return response.create_fatal_response() except ClientError as e: return response.create_error_response( code=error_const.ERROR_CODE_S3_ERROR, message=e.response['Error']['Message'], status=int(e.response['Error']['Code'])) return response.Response(message=url)