Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def _clean_single_schedule(cls, check, schedule_dict):
if not isinstance(schedule_dict, dict):
raise ConfigurationError(
'Check {0} has invalid schedule configuration: {1}'
.format(check['name'], schedule_dict),
)
try:
every = schedule_dict['every']
except KeyError:
raise ConfigurationError(
"Check {0} has invalid schedule format: {1}"
.format(check['name'], schedule_dict)
)
if isinstance(every, six.string_types):
# "every: day" shortcut
unit, every = every, 1
else:
unit = schedule_dict.get('unit')
unit = cls._clean_unit(check, unit)
def _clean_unit(check, unit):
try:
getattr(schedule.every(1), unit)
except:
raise ConfigurationError(
"Unit {0} is not one of valid options. Referenced in check {1}"
.format(unit, check['name'])
)
return unit
def fetcher_factory(conf):
"""Return initialized fetcher capable of processing given conf."""
global PROMOTERS
applicable = []
if not PROMOTERS:
PROMOTERS = load_promoters()
for promoter in PROMOTERS:
if promoter.is_applicable(conf):
applicable.append((promoter.PRIORITY, promoter))
if applicable:
best_match = sorted(applicable, reverse=True)[0][1]
return best_match(conf)
else:
raise ConfigurationError(
'No fetcher is applicable for "{0}"'.format(conf['name'])
)
def _clean_at(cls, check, at):
if at is None or cls.RE_TIME.match(at):
return at
raise ConfigurationError(
'Check {0} schedule has invalid value for "at": {1}'
.format(check['name'], at)
)
unit='seconds',
at=None
))
if "schedule" in check:
schedule_value = check['schedule']
if isinstance(schedule_value, dict):
check_schedule.append(
cls._clean_single_schedule(check, schedule_value)
)
elif isinstance(schedule_value, list):
check_schedule.extend([
cls._clean_single_schedule(check, item)
for item in schedule_value
])
else:
raise ConfigurationError(
'Check {0} has invalid schedule configuration: {1}'
.format(check['name'], check['schedule'])
)
return check_schedule