Source code for riprova.errors

# -*- coding: utf-8 -*-
from .exceptions import NotRetriableError

[docs]class ErrorWhitelist(object): """ Stores an error whitelist and provides a simple interface for whitelist update and mutation. Arguments: errors (set|list|tuple[Exception]): optional list of error exceptions classes to whitelist. Attributes: errors (list): list of whilelist errors. """ # Whitelist built-in exceptions that would be ignored by the retrier # engine. User can mutate and extend this list via class method. WHITELIST = set([ SystemExit, IndexError, ImportError, SyntaxError, ReferenceError, KeyboardInterrupt, NotRetriableError ]) def __init__(self, errors=None): self._list = set(errors if errors else ErrorWhitelist.WHITELIST.copy()) @property def errors(self): """ Sets a new error whitelist, replacing the existent one. Arguments: errors (list|tuple[Exception]): iterable containing errors to whitelist. """ return self._list @errors.setter def errors(self, errors): """ Sets whilelist errors. Raises: TypeError: if set value is not a list or tuple """ if not isinstance(errors, (list, tuple)): raise TypeError('errors must be a list or tuple') self._list = set() for err in errors: if not issubclass(err, BaseException): raise TypeError('error must be a subclass of Exception') self._list.add(err)
[docs] def add(self, *errors): """ Adds one or multiple error classes to the current whitelist. Arguments: *errors (Exception): variadic error classes to add. """ # Cache current whitelist whitelist = self._list.copy() # Delegate to attribute setter to run type validations self.errors = errors # Join whitelist with previous one self._list = whitelist | self._list
[docs] def isretry(self, error): """ Checks if a given error object is whitelisted or not. Returns: bool """ return not all([ error is not None, any(isinstance(error, err) for err in self._list), getattr(error, '__retry__', False) is False ])
[docs]class ErrorBlacklist(ErrorWhitelist): """ Provides errors blacklist used to determine those exception errors who should be retried. Implements the opposite behaviour to `ErrorWhitelist`. Arguments: errors (set|list|tuple[Exception]): optional list of error exceptions classes to blacklist. Attributes: errors (list): list of blacklist errors. """
[docs] def isretry(self, error): """ Checks if a given error object is not whitelisted. Returns: bool """ return not ErrorWhitelist.isretry(self, error)
[docs]def add_whitelist_error(*errors): """ Add additional custom errors to the global whitelist. Raises exceptions that are instances of the whitelisted errors won't be retried and they will be re-raised instead. Arguments: *errors (Exception): variadic error classes to whitelist. Usage:: riprova.add_whitelist_error(MyCustomError, AnotherError) """ [ErrorWhitelist.WHITELIST.add(error) for error in errors if issubclass(error, BaseException)]