"""Config file for lambda function.""" import os from secrets_manager.python_ext import PythonSecretsManager APPLICATION_NAME = 'lambda-fan-campaign' # AWS region configuration DEFAULT_AWS_REGION = 'us-east-1' AWS_REGION = os.environ.get('AWS_REGION', DEFAULT_AWS_REGION) # Environment setup TEST_ENVIRONMENT = 'test' DEV_ENVIRONMENT = 'dev' QA_ENVIRONMENT = 'qa' PROD_ENVIRONMENT = 'prod' ENVIRONMENT = os.environ.get('ENVIRONMENT', DEV_ENVIRONMENT) SENTRY_DSN = os.environ.get('SENTRY_DSN') # Config for Kafka DLQ. KAFKA_BOOTSTRAP_SERVERS = os.environ.get('KAFKA_BOOTSTRAP_SERVERS') KAFKA_SECURITY_PROTOCOL = os.environ.get('KAFKA_SECURITY_PROTOCOL', 'SSL') DLQ_TOPIC = os.environ.get('DLQ_TOPIC', 'dlq.lambda.campaign') # Snowflake credentials. secrets_manager_client = PythonSecretsManager( environment=ENVIRONMENT, service_name=APPLICATION_NAME) SNOWFLAKE_ACCOUNT = os.environ.get('SNOWFLAKE_ACCOUNT', 'delphi.us-east-1') SNOWFLAKE_USER = os.environ.get('SNOWFLAKE_USER') SNOWFLAKE_DATABASE = os.environ.get('SNOWFLAKE_DATABASE', 'FANSIFTER_APP_REPORTING') SNOWFLAKE_SCHEMA = os.environ.get('SNOWFLAKE_SCHEMA', 'QA_CRM_FANS') SNOWFLAKE_WAREHOUSE = os.environ.get('SNOWFLAKE_WAREHOUSE', 'QA_ETL_WH') SNOWFLAKE_TABLE = os.environ.get('SNOWFLAKE_TABLE', 'EVENT_CAMPAIGN_RAW') if ENVIRONMENT == TEST_ENVIRONMENT: # explicitly do this to avoid botocore NoCredentialsError SNOWFLAKE_PRIVATE_KEY = os.environ.get('SNOWFLAKE_PRIVATE_KEY') SNOWFLAKE_PRIVATE_KEY_PASS = os.environ.get('SNOWFLAKE_PRIVATE_KEY_PASS') else: # when ENVIRONMENT=dev or test it will get values from env variables. SNOWFLAKE_PRIVATE_KEY = secrets_manager_client.get_cred('SNOWFLAKE_PRIVATE_KEY') SNOWFLAKE_PRIVATE_KEY_PASS = secrets_manager_client.get_cred('SNOWFLAKE_PRIVATE_KEY_PASS') if not SNOWFLAKE_USER or not SNOWFLAKE_PRIVATE_KEY or not SNOWFLAKE_PRIVATE_KEY_PASS: raise Exception('Missing snowflake credentials.')