"""Helper functions for workflows.""" import base64 import os import boto3 from botocore.exceptions import ClientError import config def get_secret(secret_name): """Get secret from AWS Secrets Manager.""" full_secret_name = "{}/{}/{}".format(config.env, config.FLOW_NAME, secret_name) region_name = "us-east-1" session = boto3.session.Session() client = session.client( service_name='secretsmanager', region_name=region_name) try: get_secret_value_response = client.get_secret_value( SecretId=full_secret_name) except ClientError as error: if error.response['Error']['Code'] in ['DecryptionFailureException', 'InternalServiceErrorException', 'InvalidParameterException', 'InvalidRequestException', 'ResourceNotFoundException']: raise error else: if 'SecretString' in get_secret_value_response: return get_secret_value_response['SecretString'].\ replace(secret_name, '').replace('"', '').replace(":", '').\ replace("{", '').replace("}", '') return base64.b64decode(get_secret_value_response['SecretBinary']) def get_youtube_api_credentials(): """Return YouTube Data API credentials.""" client_id = [get_secret('YOUTUBE_CLINET_ID_1'), get_secret('YOUTUBE_CLINET_ID_2'), get_secret('YOUTUBE_CLINET_ID_3'), get_secret('YOUTUBE_CLINET_ID_4'), ] config.youtube['client_id'] = client_id return config.youtube