Module pynom.pynom

Expand source code
import collections
import dataclasses
import datetime
import traceback
import types
import typing


@dataclasses.dataclass
class ExceptionInfo:
    """A dataclass containing an exception, traceback, and approximate timestamp"""

    exception: Exception
    traceback: types.TracebackType
    time_stamp: datetime.datetime

    def __str__(self):
        ret_str = f"{type(self.exception).__name__}: thrown at {self.time_stamp}\n"
        ret_str += ''.join(traceback.format_exception(type(self.exception), self.exception, self.traceback))
        return ret_str


class CombinedException(Exception):
    """An exception that will be raised containing the currently eatten exceptions of the same type"""

    def __init__(self, exception_infos: typing.List[ExceptionInfo]):
        self.exception_infos = exception_infos

    def __str__(self):
        return "CombinedException:\n  " + (
            "\n----------\n".join([str(a) for a in self.exception_infos])
        ).replace("\n", "\n  ").rstrip(" ")


class PyNom:
    #: A special variable that corresponds with an exception type leading to all exceptions being eatten even if not listed in exception_types_to_eat
    ALL_EXCEPTIONS = "ALL_EXCEPTIONS"

    def __init__(
        self,
        exception_types_to_eat: list,
        max_to_eat_before_throw_up: int,
        throw_up_action: typing.Union[typing.Callable, None] = None,
        side_dish_action: typing.Union[typing.Callable, None] = None,
        digest_time: typing.Union[datetime.timedelta, None] = None,
        digest_action: typing.Union[typing.Callable, None] = None,
    ):
        """Instantiate a PyNom object

        Args:
            exception_types_to_eat: A list of exception types that PyNom should eat.
                Note that if PyNom.ALL_EXCEPTIONS is given, all exceptions will be eaten.
                    Note 2: Passing the max_to_eat_before_throw_up for a given exception type will still lead to a throw up.
            max_to_eat_before_throwing_up: The max exceptions of a given type to eat before throwing up
            throw_up_action: A callable to call when throwing up. The callable should take one arg (a CombinedException).
                If None is given, raise the CombinedException.
            side_dish_action: A callable to call when attempting to eat an exception. Note that performing the side_dish_action does
                not guarantee that a throw_up_action will not be called soon after. The callable should take one arg: an ExceptionInfo.
            digest_time: A timedelta that it should take a caught exception to take to 'digest'. If an exception is digested, it will be
                forgotten and not thrown up (as in it will no longer count towards the max_to_eat_before_throwing_up). By default None is
                given which means no Exception will be 'digested'/forgotten.
            digest_action: A callable to call when digesting an exception. If given, the callable should take one arg: an ExceptionInfo.
        """
        self.exception_types_to_eat = (
            exception_types_to_eat
            if isinstance(exception_types_to_eat, (list, set))
            else [exception_types_to_eat]
        )
        self.max_to_eat_before_throw_up = max_to_eat_before_throw_up
        self.throw_up_action = throw_up_action
        self.side_dish_action = side_dish_action
        self.digest_time = digest_time
        self.digest_action = digest_action
        self._exception_information = collections.defaultdict(list)

    def _is_eating_all_exceptions(self):
        return self.ALL_EXCEPTIONS in self.exception_types_to_eat

    def __enter__(self):
        return self

    def _check_and_perform_digestion(self, typ: typing.Type, now: datetime.datetime):
        """
        Checks the currently saved exceptions for the given type and removes once that expired based
        off the digest_time. Will also call the digest_action if given.
        """
        if self.digest_time is not None:
            new_ex_infos = []
            for ex_info in self._exception_information[typ]:
                ex_info = typing.cast(ExceptionInfo, ex_info)
                if (now - ex_info.time_stamp) >= self.digest_time:
                    if self.digest_action is not None:
                        self.digest_action(ex_info)
                else:
                    new_ex_infos.append(ex_info)

            self._exception_information[typ] = new_ex_infos

    def __exit__(self, typ, value, traceback):
        if (
            typ not in self.exception_types_to_eat
            and not self._is_eating_all_exceptions()
        ) or value is None:
            # about to raise a not eaten exception OR there is no exxception
            return

        now = datetime.datetime.now()
        ex_info = ExceptionInfo(value, traceback, now)
        self._exception_information[typ].append(ex_info)

        # check all current exception_information for if it should be digested
        self._check_and_perform_digestion(typ, now)

        if self.side_dish_action is not None:
            self.side_dish_action(ex_info)

        if len(self._exception_information[typ]) > self.max_to_eat_before_throw_up:
            # define class here so it subclasses the type of the raised exception
            class FullCombinedException(CombinedException, typ):
                pass

            raise_list = self._exception_information[typ]
            self._exception_information[typ] = []

            ex = FullCombinedException(raise_list)
            if self.throw_up_action is None:
                raise ex
            else:
                self.throw_up_action(ex)

        # if we get here, do not throw anything
        return True

Classes

class CombinedException (exception_infos: List[ExceptionInfo])

An exception that will be raised containing the currently eatten exceptions of the same type

Expand source code
class CombinedException(Exception):
    """An exception that will be raised containing the currently eatten exceptions of the same type"""

    def __init__(self, exception_infos: typing.List[ExceptionInfo]):
        self.exception_infos = exception_infos

    def __str__(self):
        return "CombinedException:\n  " + (
            "\n----------\n".join([str(a) for a in self.exception_infos])
        ).replace("\n", "\n  ").rstrip(" ")

Ancestors

  • builtins.Exception
  • builtins.BaseException
class ExceptionInfo (exception: Exception, traceback: traceback, time_stamp: datetime.datetime)

A dataclass containing an exception, traceback, and approximate timestamp

Expand source code
class ExceptionInfo:
    """A dataclass containing an exception, traceback, and approximate timestamp"""

    exception: Exception
    traceback: types.TracebackType
    time_stamp: datetime.datetime

    def __str__(self):
        ret_str = f"{type(self.exception).__name__}: thrown at {self.time_stamp}\n"
        ret_str += ''.join(traceback.format_exception(type(self.exception), self.exception, self.traceback))
        return ret_str

Class variables

var exception : Exception
var time_stamp : datetime.datetime
var traceback : traceback
class PyNom (exception_types_to_eat: list, max_to_eat_before_throw_up: int, throw_up_action: Optional[Callable] = None, side_dish_action: Optional[Callable] = None, digest_time: Optional[datetime.timedelta] = None, digest_action: Optional[Callable] = None)

Instantiate a PyNom object

Args

exception_types_to_eat
A list of exception types that PyNom should eat. Note that if PyNom.ALL_EXCEPTIONS is given, all exceptions will be eaten. Note 2: Passing the max_to_eat_before_throw_up for a given exception type will still lead to a throw up.
max_to_eat_before_throwing_up
The max exceptions of a given type to eat before throwing up
throw_up_action
A callable to call when throwing up. The callable should take one arg (a CombinedException). If None is given, raise the CombinedException.
side_dish_action
A callable to call when attempting to eat an exception. Note that performing the side_dish_action does not guarantee that a throw_up_action will not be called soon after. The callable should take one arg: an ExceptionInfo.
digest_time
A timedelta that it should take a caught exception to take to 'digest'. If an exception is digested, it will be forgotten and not thrown up (as in it will no longer count towards the max_to_eat_before_throwing_up). By default None is given which means no Exception will be 'digested'/forgotten.
digest_action
A callable to call when digesting an exception. If given, the callable should take one arg: an ExceptionInfo.
Expand source code
class PyNom:
    #: A special variable that corresponds with an exception type leading to all exceptions being eatten even if not listed in exception_types_to_eat
    ALL_EXCEPTIONS = "ALL_EXCEPTIONS"

    def __init__(
        self,
        exception_types_to_eat: list,
        max_to_eat_before_throw_up: int,
        throw_up_action: typing.Union[typing.Callable, None] = None,
        side_dish_action: typing.Union[typing.Callable, None] = None,
        digest_time: typing.Union[datetime.timedelta, None] = None,
        digest_action: typing.Union[typing.Callable, None] = None,
    ):
        """Instantiate a PyNom object

        Args:
            exception_types_to_eat: A list of exception types that PyNom should eat.
                Note that if PyNom.ALL_EXCEPTIONS is given, all exceptions will be eaten.
                    Note 2: Passing the max_to_eat_before_throw_up for a given exception type will still lead to a throw up.
            max_to_eat_before_throwing_up: The max exceptions of a given type to eat before throwing up
            throw_up_action: A callable to call when throwing up. The callable should take one arg (a CombinedException).
                If None is given, raise the CombinedException.
            side_dish_action: A callable to call when attempting to eat an exception. Note that performing the side_dish_action does
                not guarantee that a throw_up_action will not be called soon after. The callable should take one arg: an ExceptionInfo.
            digest_time: A timedelta that it should take a caught exception to take to 'digest'. If an exception is digested, it will be
                forgotten and not thrown up (as in it will no longer count towards the max_to_eat_before_throwing_up). By default None is
                given which means no Exception will be 'digested'/forgotten.
            digest_action: A callable to call when digesting an exception. If given, the callable should take one arg: an ExceptionInfo.
        """
        self.exception_types_to_eat = (
            exception_types_to_eat
            if isinstance(exception_types_to_eat, (list, set))
            else [exception_types_to_eat]
        )
        self.max_to_eat_before_throw_up = max_to_eat_before_throw_up
        self.throw_up_action = throw_up_action
        self.side_dish_action = side_dish_action
        self.digest_time = digest_time
        self.digest_action = digest_action
        self._exception_information = collections.defaultdict(list)

    def _is_eating_all_exceptions(self):
        return self.ALL_EXCEPTIONS in self.exception_types_to_eat

    def __enter__(self):
        return self

    def _check_and_perform_digestion(self, typ: typing.Type, now: datetime.datetime):
        """
        Checks the currently saved exceptions for the given type and removes once that expired based
        off the digest_time. Will also call the digest_action if given.
        """
        if self.digest_time is not None:
            new_ex_infos = []
            for ex_info in self._exception_information[typ]:
                ex_info = typing.cast(ExceptionInfo, ex_info)
                if (now - ex_info.time_stamp) >= self.digest_time:
                    if self.digest_action is not None:
                        self.digest_action(ex_info)
                else:
                    new_ex_infos.append(ex_info)

            self._exception_information[typ] = new_ex_infos

    def __exit__(self, typ, value, traceback):
        if (
            typ not in self.exception_types_to_eat
            and not self._is_eating_all_exceptions()
        ) or value is None:
            # about to raise a not eaten exception OR there is no exxception
            return

        now = datetime.datetime.now()
        ex_info = ExceptionInfo(value, traceback, now)
        self._exception_information[typ].append(ex_info)

        # check all current exception_information for if it should be digested
        self._check_and_perform_digestion(typ, now)

        if self.side_dish_action is not None:
            self.side_dish_action(ex_info)

        if len(self._exception_information[typ]) > self.max_to_eat_before_throw_up:
            # define class here so it subclasses the type of the raised exception
            class FullCombinedException(CombinedException, typ):
                pass

            raise_list = self._exception_information[typ]
            self._exception_information[typ] = []

            ex = FullCombinedException(raise_list)
            if self.throw_up_action is None:
                raise ex
            else:
                self.throw_up_action(ex)

        # if we get here, do not throw anything
        return True

Class variables

var ALL_EXCEPTIONS

A special variable that corresponds with an exception type leading to all exceptions being eatten even if not listed in exception_types_to_eat