"""Config file for lambda function.""" import os from lambdacommon import common_config from secrets_manager.lambda_ext import LambdaSecretsManager logger = common_config.logger ENVIRONMENT = os.environ.get('ENVIRONMENT', common_config.ENVIRONMENT) JENKINS_URL = os.environ.get( 'JENKINS_URL', 'https://scheduler.theorchard.io', ) JENKINS_USER = os.environ.get('JENKINS_USER') JENKINS_API_TOKEN_NAME = 'JENKINS_API_TOKEN' JENKINS_API_TOKEN_SECRETS_SERVICE = 'swf-shared-jenkins' # list of environments to trigger Jenkins jobs JENKINS_ENVIRONMENTS = os.environ.get( 'JENKINS_TRIGGER_ENVIRONMENTS', ENVIRONMENT ).split(',') def get_jenkins_token(): """Get Jenkins API token from environment variable or secrets manager.""" logger.info('Reading Jenkins token from secrets manager') # local - should read from ENV # dev - should read from secrets manager is_local = ENVIRONMENT == 'local' secrets_manager = LambdaSecretsManager( environment='dev' if is_local else ENVIRONMENT, service_name=JENKINS_API_TOKEN_SECRETS_SERVICE, force_remote=not is_local, ) secret_text = secrets_manager.get_cred(JENKINS_API_TOKEN_NAME) return secret_text JENKINS_JOB_AMAZON_DATAPULSE = 'swf-amazon-datapulse-exec' JENKINS_ARGS_REPORT_DATE = { 'ENVIRONMENT': '{environment}', 'REPORT': '{report_type}', 'CONTEXT_DATE': '{report_date}', 'PARTITION': '{partition}', } JENKINS_ARGS_YMD = { 'ENVIRONMENT': '{environment}', 'REPORT': '{report_type}', 'CONTEXT_DATE': '{Year}-{Month}-{Day}', 'PARTITION': '{partition}', } REPORTS = { 'fraud_report': { 'jenkins_job': JENKINS_JOB_AMAZON_DATAPULSE, 'jenkins_args': JENKINS_ARGS_REPORT_DATE, }, 'daily_play_events': { 'jenkins_job': JENKINS_JOB_AMAZON_DATAPULSE, 'jenkins_args': JENKINS_ARGS_YMD, }, 'daily_customer_metadata': { 'jenkins_job': JENKINS_JOB_AMAZON_DATAPULSE, 'jenkins_args': JENKINS_ARGS_YMD, }, 'daily_playlist_metadata': { 'jenkins_job': JENKINS_JOB_AMAZON_DATAPULSE, 'jenkins_args': JENKINS_ARGS_YMD, }, }