How to use the pypyr.utils.poll.while_until_true 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
sleep = context.get_formatted_as_type(self.sleep, out_type=float)
        if self.max:
            max = context.get_formatted_as_type(self.max, out_type=int)

            logger.info("retry decorator will try %d times at %ss "
                        "intervals.", max, sleep)
        else:
            max = None
            logger.info("retry decorator will try indefinitely at %ss "
                        "intervals.", sleep)

        # this will never be false. because on counter == max,
        # exec_iteration raises an exception, breaking out of the loop.
        # pragma because cov doesn't know the implied else is impossible.
        # unit test cov is 100%, though.
        if poll.while_until_true(interval=sleep,
                                 max_attempts=max)(
                self.exec_iteration)(context=context,
                                     step_method=step_method
                                     ):  # pragma: no cover
            logger.debug("retry loop complete, reporting success.")

        logger.debug("retry loop done")

        logger.debug("done")
github pypyr / pypyr-cli / pypyr / dsl.py View on Github external
if max < 1:
                logger.info(
                    "max %s is %s. while only runs when max > 0.",
                    self.max, max)
                logger.debug("done")
                return

            if self.stop is None:
                logger.info("while decorator will loop %s times at "
                            "%ss intervals.", max, sleep)
            else:
                logger.info("while decorator will loop %s times, or "
                            "until %s evaluates to True at "
                            "%ss intervals.", max, self.stop, sleep)

        if not poll.while_until_true(interval=sleep,
                                     max_attempts=max)(
                self.exec_iteration)(context=context,
                                     step_method=step_method):
            # False means loop exhausted and stop never eval-ed True.
            if error_on_max:
                logger.error("exhausted %s iterations of while loop, "
                             "and errorOnMax is True.", max)
                if self.stop and max:
                    raise LoopMaxExhaustedError("while loop reached "
                                                f"{max} and {self.stop} "
                                                "never evaluated to True.")
                else:
                    raise LoopMaxExhaustedError(f"while loop reached {max}.")
            else:
                if self.stop and max:
                    logger.info(