def lambda_handler()

in sam-app/lambda_functions/sfGenerateAudioRecordingStreamingURL.py [0:0]

• 27 lines of code
• 6 McCabe index (conditional complexity)

def lambda_handler(event, context):
    if 'recordingPath' not in event or not event['recordingPath'] or event['recordingPath'] == 'null':
        logger.info("No recordingPath in event; returning.")
        return None
    # retrieve secrets
    logger.info("Retrieving cloudfront credentials")
    session = boto3.session.Session()
    client = session.client(service_name='secretsmanager')
    sf_credentials_secrets_manager_arn = get_arg(os.environ,
            'SF_CREDENTIALS_SECRETS_MANAGER_ARN')
    secrets = json.loads(client.get_secret_value(SecretId=sf_credentials_secrets_manager_arn)['SecretString'])
    private_key = secrets['CloudFrontPrivateKey']
    access_key_id = secrets['CloudFrontAccessKeyID']
    logger.info("Cloudfront credentials retrieved")

    # construct url to audio recording
    recordingPath = event['recordingPath'] # need to remove bucket name, connect dir from path
    if("/connect/" in recordingPath):
        recordingPath = "connect/" + recordingPath.split("/connect/", 1)[1]
    elif("/Analysis/" in recordingPath):
        recordingPath = "Analysis/" + recordingPath.split("/Analysis/", 1)[1]
    cloudfront_domain = get_arg(os.environ, 'CLOUDFRONT_DISTRIBUTION_DOMAIN_NAME')
    url = 'https://' + cloudfront_domain + '/' + recordingPath
    logger.info('Unsigned audio recording url: %s' % url)

    # sign url
    expire_date = datetime.datetime.utcnow() + datetime.timedelta(minutes=60)
    cloudfront_signer = CloudFrontSigner(access_key_id, rsa_signer(private_key))
    signed_url = cloudfront_signer.generate_presigned_url(
        url, date_less_than=expire_date)
    logger.info('Signed audio recording url: %s' % signed_url)
    return signed_url