import os from tadas.platform import logging as logging_utils logger = logging_utils.get_file_logger(__file__) # values can be overridden by environment variables DEFAULTS = { 'ENVIRONMENT': 'dev', # todo: remove from configurable options. Maybe in constants ? 'REPORT_DATE': '', # desired report_date in YYYY-MM-DD 'MODEL_VERSION': 'initial', 'REQUIRED_DSPS': [], # check for only these DSPs for availability 'MAKE_TASKS': ['inference', 'publish'], # tasks from Makefile to run. Can add "clean" 'DAYS_BACK': 5, # number of days back from now to lookup for available data 'SOURCE_SCHEMA': 'facts.prod.', # if empty then use default connection schema 'SOURCE_DB_TYPE': 'reporting', 'USE_CACHE': False, # cache SQL results on disk (only for development) 'CACHE_TTL_MINUTES': 60 * 24, # default cache TTL 'SAVE_DAYS_TRENDING': False, 'TADAS_UPDATE_DYNAMODB_STATUS': False, 'TADAS_TRIGGER_JENKINS': False, 'TRENDING_DAYS_LOOKBACK_PERIOD_DAYS': 70, 'TRENDING_DAYS_UPPER_LIMIT': 30, # do not publish tracks trending for more than this limit 'LAUNCHER_RELEASE_LOCK': False, # force to release existing lock in launcher 'FORCE_PUBLISH': False, # force to sync to target tables 'FRESHNESS_THRESHOLD_HOURS': 74, # must be synced with freshness interval in https://sonymusic-pde.datadoghq.com/notebook/13556459/tadas-etl } def get(setting: str): """ Get configuration setting value. :raises KeyError: If the setting name is not recognized. """ try: default_value = DEFAULTS[setting] except KeyError: raise KeyError(f'Unknown configuration setting: {setting}') value_type = type(default_value) env_value = os.environ.get(setting) if env_value is None or env_value == '': if value_type == list: default_value = default_value.copy() result = default_value elif value_type == bool: if env_value.lower() in ('y', 'yes', 'true'): result = True elif env_value.lower() in ('n', 'no', 'false'): result = False else: raise ValueError(f'Invalid boolean value: {env_value}') elif value_type == int: result = int(env_value) elif value_type == list: result = env_value.split(',') elif value_type == str: result = env_value else: raise ValueError(f'Unsupported type: {value_type}') logger.info(f'{setting}={result}') return result