Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def save_error(self, context, exception, swallowed):
"""Append step's exception information to the context.
Append the[on_error] dictionary to the context. This will append to
existing `runErrors` values if `runErrors` are already in there.
Args:
context: (pypyr.context.Context) The pypyr context. This arg will
mutate - after method execution will contain the new
updated context.
exception: (Exception) The error detected during step execution.
swallowed: (bool) Whether exception was swallowed or not.
"""
failure = {
'name': get_error_name(exception),
'description': str(exception),
'customError': context.get_formatted_iterable(
self.on_error
) if self.on_error else {},
'line': self.line_no,
'col': self.line_col,
'step': self.name,
'exception': exception,
'swallowed': swallowed,
}
run_errors = context.setdefault('runErrors', [])
run_errors.append(failure)
except Exception as exc_info:
if isinstance(exc_info, HandledError):
exc_info = exc_info.__cause__
else:
# prevent already logged err logging twice.
self.save_error(
context=context,
exception=exc_info,
swallowed=swallow_me
)
if swallow_me:
logger.error(
"%s Ignoring error because swallow "
"is True for this step.\n"
"%s: %s",
self.name, get_error_name(exc_info), exc_info
)
else:
if self.line_no:
logger.error(
"Error while running step %s "
"at pipeline yaml line: %d, col: %d",
self.name, self.line_no, self.line_col
)
else:
logger.error(
"Error while running step %s", self.name
)
raise exc_info
else:
logger.info(
"%s not running because skip is True.", self.name)
result = True
except (ControlOfFlowInstruction, Stop):
# Control-of-Flow/Stop are instructions to go somewhere
# else, not errors per se.
raise
except Exception as ex_info:
if self.max:
if counter == self.max:
logger.debug("retry: max %s retries exhausted. "
"raising error.", counter)
# arguably shouldn't be using errs for control of flow.
# but would lose the err info if not, so lesser of 2 evils.
raise
if self.stop_on or self.retry_on:
error_name = get_error_name(ex_info)
if self.stop_on:
formatted_stop_list = context.get_formatted_iterable(
self.stop_on)
if error_name in formatted_stop_list:
logger.error("%s in stopOn. Raising error "
"and exiting retry.", error_name)
raise
else:
logger.debug("%s not in stopOn. Continue.", error_name)
if self.retry_on:
formatted_retry_list = context.get_formatted_iterable(
self.retry_on)
if error_name not in formatted_retry_list:
logger.error("%s not in retryOn. Raising "
"error and exiting retry.", error_name)