in source/lambda/configuration/config_admin.py [0:0]
def update_config(self, **settings):
"""
Updates configuration, validates new values
:param settings: settings values
:return: updated values
"""
valid_attributes = [configuration.METRICS,
configuration.CROSS_ACCOUNT_ROLES,
configuration.DEFAULT_TIMEZONE,
configuration.REGIONS,
configuration.SCHEDULE_LAMBDA_ACCOUNT,
configuration.TAGNAME,
configuration.TRACE,
configuration.ENABLE_SSM_MAINTENANCE_WINDOWS,
ConfigAdmin.TYPE_ATTR,
configuration.SCHEDULED_SERVICES,
configuration.SCHEDULE_CLUSTERS,
configuration.CREATE_RDS_SNAPSHOT,
configuration.STARTED_TAGS,
configuration.STOPPED_TAGS]
checked_settings = {}
for attr in settings:
if attr in [ConfigAdmin.TYPE_ATTR, configuration.NAME]:
continue
# only valid fields
if attr not in valid_attributes:
raise ValueError(ERR_UPDATE_UNKNOWN_PARAMETER.format(attr))
# remove None fields
if settings[attr] is None:
continue
# remove empty strings
if len(str(settings[attr])) == 0:
continue
# make sure these fields are set as sets
if attr in [configuration.REGIONS, configuration.CROSS_ACCOUNT_ROLES, configuration.SCHEDULED_SERVICES]:
temp = self._ensure_set(settings[attr])
if len(settings[attr]) > 0:
checked_settings[attr] = temp
continue
# make sure these fields are valid booleans
if attr in [configuration.METRICS,
configuration.TRACE,
configuration.ENABLE_SSM_MAINTENANCE_WINDOWS,
configuration.SCHEDULE_LAMBDA_ACCOUNT,
configuration.CREATE_RDS_SNAPSHOT,
configuration.SCHEDULE_CLUSTERS]:
bool_value = self._ensure_bool(settings[attr])
if bool_value is None:
raise ValueError(ERR_UPDATE_INVALID_BOOL_PARAM.format(settings[attr], attr))
checked_settings[attr] = bool_value
continue
# validate timezone
if attr == configuration.DEFAULT_TIMEZONE:
default_tz = settings[configuration.DEFAULT_TIMEZONE]
if not SchedulerConfigBuilder.is_valid_timezone(default_tz):
raise ValueError(
ERR_UPDATE_INVALID_TZ_PARAMETER.format(default_tz, configuration.DEFAULT_TIMEZONE))
checked_settings[attr] = default_tz
continue
checked_settings[attr] = settings[attr]
if configuration.TAGNAME not in settings:
raise ValueError(ERR_UPDATE_TAGNAME_EMPTY)
for service in settings.get(configuration.SCHEDULED_SERVICES, []):
if service not in ConfigAdmin.SUPPORTED_SERVICES:
raise ValueError(ERR_UPDATE_UNKNOWN_SERVICE.format(service))
# keys for config item
checked_settings[ConfigAdmin.TYPE_ATTR] = "config"
checked_settings[configuration.NAME] = "scheduler"
self._table.put_item_with_retries(Item=checked_settings)
return ConfigAdmin._for_output(checked_settings)