Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
# remove=True)
instance = get_setting_value(message,
'instance_field',
default=HOSTNAME,
remove=True)
logger.debug(instance)
device = get_setting_value(message,
'device_field',
remove=True)
# get timestamp
try:
timestamp = get_setting_value(message,
'timestamp_field',
remove=True)
timestamp = [timestamp]
except ListNotAllowedError as e:
logger.debug(e)
timestamp = get_setting_value(message,
'timestamp_field',
remove=True,
allow_list=True)
except Exception as e:
logger.warn(e)
sys.exit(1)
logger.debug(timestamp)
# get data
data = get_data_values(timestamp, message)
# hand off
for timestamp, report_data in data.items():
# check if this is within the time range
# continue traversing?
if next_fields is None:
# some error, but return the value
return next_value
elif len(next_fields) == 0:
# final value in the list to walk down
return next_value
elif isinstance(next_value, (list, set, tuple)):
# we've reached an early terminal point, which may or may not be ok
if allow_list:
return json_gather_list_values(
next_value,
next_fields,
remove=remove)
else:
raise ListNotAllowedError('encountered list or set in json when not allowed')
elif isinstance(next_value, dict):
# there's more tree to walk down
return _get_json_field_helper(
json.loads(json.dumps(next_value)),
next_fields,
allow_list=allow_list,
remove=remove)
else:
# catch-all
return ''