"""Interface for s3.""" import json import boto3 from botocore.exceptions import ClientError from sound_recordings import config s3_client = boto3.client('s3') def get_sound_recording_version(sound_recording_id, version_id): """Retrieve sound recording data by version. Args: sound_recording_id (str): sound recording id version id (str): version id Returns: dict: data if retrieved else None """ data = None try: params = { 'Bucket': str(config.VERSION_BUCKET), 'Key': f'{config.VERSION_BUCKET_KEY_PREFIX}{sound_recording_id}' } if version_id != 'latest': params['VersionId'] = version_id response = s3_client.get_object(**params) data = { 'version_id': response['VersionId'], 'creation_date': response['LastModified'], 'data': json.loads(response['Body'].read()) } except ClientError as e: if e.response['Error']['Code'] == 'NoSuchKey' \ or e.response['Error']['Message'] == 'Invalid version id specified': return data else: raise e return data def get_signed_url(path): """Retrieve signed url by path. Args: path (str): s3 path Returns: str: signed url """ data = None try: params = { 'Bucket': str(config.DELIVERY_BUCKET), 'Key': f'{config.DELIVERY_BUCKET_PREFIX}{path}' } expiration = 86400000 data = s3_client.generate_presigned_url( 'get_object', Params=params, ExpiresIn=expiration) except ClientError as e: if e.response['Error']['Code'] == 'NoSuchKey': return data else: raise e return data