Python Exception Monitor

Question

I have built a data pipeline in Python (3.12). It typically processes hundreds of thousands of files and runs for several days. A particular pipeline function calls external web APIs that sometimes fail. I expect some exceptions, and I want to record them. However, if the exception frequency is above a threshold, I do want to abort immediately.

I implemented an exception class that I use to decorate the function above.

Implementation:

class ExceptionMonitor:
    """ Monitor the number of exceptions raised and the number of calls made to a function. """
    def __init__(self, error_rate_threshold: float, min_calls: int):
        """
        :param error_rate_threshold: The error rate threshold above which an exception is raised.
        :param min_calls: The minimum number of calls before the error rate is calculated.
        """
        self.error_rate_threshold = error_rate_threshold
        self.min_calls = min_calls
        self.nb_exceptions = 0
        self.nb_calls = 0
        self.error_rate = 0

    def _save_exception(self):
        with open("exceptions.log", "a") as f:
            f.write(f"{datetime.now()}\n")
            f.write(traceback.format_exc())
            f.write("\n\n")


    def __call__(self, func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            try:
                return func(*args, **kwargs)
            except Exception:
                self.nb_exceptions += 1
                if self.nb_calls >= self.min_calls:
                    self.error_rate = self.nb_exceptions / self.nb_calls
                    if self.error_rate > self.error_rate_threshold:
                        raise
                self._save_exception()
            finally:
                self.nb_calls += 1
        return wrapper

Usage:

@ExceptionMonitor(error_rate_threshold=0.03, min_calls=100)
def some_func():
   call_web_api()

Is this code idiomatic?

Answer 1

Your code looks fine to me.

For some nitpicks:

1. pass the exception to _save_exception to ensure you're interacting with the correct error, and
2. pass the path to the log file to the class, or change to a logging.Logger, to make logging to a different location easier to change.

Answer 2

It's likely to be irrelevant for most parameters, but I think that self.nb_calls += 1 belongs in the try block before calling the function rather than a finally block after.

To take a small example, consider @ExceptionMonitor(error_rate_threshold=0.5, min_calls=2) and function results like [success, success, exception]. I would argue that this does not clear the 50% threshold, however the code as written would say that it does.

This would also address @toolic's note about division by zero.

Answer 3

Since the code does a division calculation:

self.error_rate = self.nb_exceptions / self.nb_calls

it would be good to add a check that the divisor (nb_calls) is not 0. It is initialized to 0 by __init__:

self.nb_calls = 0

You might want to check that min_calls is greater than 0 as well, unless you want to support that case. If you do support 0, it might be worth explicitly documenting that.

Answer 4

One issue that I identified later:

The error rate calculation is cumulative, so error bursts will be missed. For example, if the error threshold is 0.05 (5%) and min_calls is 10 and there are 8 consecutive exceptions from the 20th to the 27th call, the cumulative error rate will be 8/27 which is around 3%. Although technically correct, it's more useful to calculate the error rate within a sliding window, like this:

class ExceptionMonitor:
    """Monitor the number of exceptions raised and the number of calls made to a function."""

    def __init__(
        self, error_rate_threshold: float, min_calls: int, window_sz: int = 100
    ):
        """
        :param error_rate_threshold: The error rate threshold in [0,1), above which an exception is raised.
        :param min_calls: The minimum number of calls before the error rate is calculated.
        :param window_sz: The size of the sliding window for calculating exception rate.

        """
        self.error_rate_threshold = error_rate_threshold
        self.min_calls = min_calls
        self.exceptions_log = "exceptions.log"
        if os.path.exists(self.exceptions_log):
            os.remove(self.exceptions_log)

        if window_sz < min_calls:
            raise ValueError(
                f"Window size {window_sz} must be greater than or equal to min_calls {min_calls}."
            )
        self.window_sz = window_sz
        self.recent_outcomes = deque(maxlen=window_sz)

    def _error_rate_exceeded(self):
        if len(self.recent_outcomes) < self.min_calls:
            return False
        return (
            sum(self.recent_outcomes) / len(self.recent_outcomes)
        ) > self.error_rate_threshold

    def _save_exception(self):
        with open("exceptions.log", "a") as f:
            f.write(f"{datetime.now()}\n")
            f.write(traceback.format_exc())
            f.write("\n\n")

    def __call__(self, func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            try:
                result = func(*args, **kwargs)
                self.recent_outcomes.append(False)
                return result
            except Exception:
                self.recent_outcomes.append(True)
                if self._error_rate_exceeded():
                    raise
                self._save_exception()

        return wrapper