How to use the pypyr.errors.get_error_name function in pypyr

To help you get started, we’ve selected a few pypyr examples, based on popular ways it is used in public projects.

Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.

github pypyr / pypyr-cli / pypyr / dsl.py View on Github external
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)
github pypyr / pypyr-cli / pypyr / dsl.py View on Github external
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)
github pypyr / pypyr-cli / pypyr / dsl.py View on Github external
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)