from abc import abstractmethod from typing import TypeVar, Callable, Any CAPTURE_EXCEPTIONS = (Exception, ) _ValueType = TypeVar('_ValueType') def _to_exception(exception) -> Callable[[Any], Exception]: """ Overview: Convert exception to callable exception. Arguments: - exception (:obj:`Exception`): The exception to be converted. """ if hasattr(exception, '__call__'): return exception elif isinstance(exception, Exception): return lambda v: exception elif isinstance(exception, str): return lambda v: ValueError(exception) else: raise TypeError( 'Unknown type of exception, func, exception or str expected but {actual} found.'.format( actual=repr(type(exception).__name__) ) ) def _to_loader(value) -> 'ILoaderClass': """ Overview: Convert value to loader. Arguments: - value (:obj:`Any`): The value to be converted. """ if isinstance(value, ILoaderClass): return value elif isinstance(value, tuple): if len(value) == 2: _predict, _exception = value _load = None elif len(value) == 3: _predict, _load, _exception = value else: raise ValueError('Tuple\'s length should be 2 or 3, but {actual} found.'.format(actual=repr(len(value)))) _exception = _to_exception(_exception) def _load_tuple(value_): if not _predict(value_): raise _exception(value_) return (_load or (lambda v: v))(value_) return _to_loader(_load_tuple) elif isinstance(value, type): def _load_type(value_): if not isinstance(value_, value): raise TypeError( 'type not match, {expect} expected but {actual} found'.format( expect=repr(value.__name__), actual=repr(type(value_).__name__) ) ) return value_ return _to_loader(_load_type) elif hasattr(value, '__call__'): class _Loader(ILoaderClass): def _load(self, value_): return value(value_) return _Loader() elif isinstance(value, bool): return _to_loader((lambda v: value, ValueError('assertion false'))) elif value is None: return _to_loader( ( lambda v: v is None, lambda v: TypeError('type not match, none expected but {actual} found'.format(actual=repr(type(v).__name__))) ) ) else: return _to_loader(lambda v: value) Loader = _to_loader def _reset_exception(loader, eg: Callable[[Any, Exception], Exception]): """ Overview: Reset exception of loader. """ loader = Loader(loader) def _load(value): try: return loader(value) except CAPTURE_EXCEPTIONS as err: raise eg(value, err) return Loader(_load) class ILoaderClass: """ Overview: Base class of loader. Interfaces: ``__init__``, ``_load``, ``load``, ``check``, ``__call__``, ``__and__``, ``__or__``, ``__rshift__`` """ @abstractmethod def _load(self, value: _ValueType) -> _ValueType: """ Overview: Load the value. Arguments: - value (:obj:`_ValueType`): The value to be loaded. """ raise NotImplementedError def __load(self, value: _ValueType) -> _ValueType: """ Overview: Load the value. Arguments: - value (:obj:`_ValueType`): The value to be loaded. """ return self._load(value) def __check(self, value: _ValueType) -> bool: """ Overview: Check whether the value is valid. Arguments: - value (:obj:`_ValueType`): The value to be checked. """ try: self._load(value) except CAPTURE_EXCEPTIONS: return False else: return True def load(self, value: _ValueType) -> _ValueType: """ Overview: Load the value. Arguments: - value (:obj:`_ValueType`): The value to be loaded. """ return self.__load(value) def check(self, value: _ValueType) -> bool: """ Overview: Check whether the value is valid. Arguments: - value (:obj:`_ValueType`): The value to be checked. """ return self.__check(value) def __call__(self, value: _ValueType) -> _ValueType: """ Overview: Load the value. Arguments: - value (:obj:`_ValueType`): The value to be loaded. """ return self.__load(value) def __and__(self, other) -> 'ILoaderClass': """ Overview: Combine two loaders. Arguments: - other (:obj:`ILoaderClass`): The other loader. """ def _load(value: _ValueType) -> _ValueType: self.load(value) return Loader(other).load(value) return Loader(_load) def __rand__(self, other) -> 'ILoaderClass': """ Overview: Combine two loaders. Arguments: - other (:obj:`ILoaderClass`): The other loader. """ return Loader(other) & self def __or__(self, other) -> 'ILoaderClass': """ Overview: Combine two loaders. Arguments: - other (:obj:`ILoaderClass`): The other loader. """ def _load(value: _ValueType) -> _ValueType: try: return self.load(value) except CAPTURE_EXCEPTIONS: return Loader(other).load(value) return Loader(_load) def __ror__(self, other) -> 'ILoaderClass': """ Overview: Combine two loaders. Arguments: - other (:obj:`ILoaderClass`): The other loader. """ return Loader(other) | self def __rshift__(self, other) -> 'ILoaderClass': """ Overview: Combine two loaders. Arguments: - other (:obj:`ILoaderClass`): The other loader. """ def _load(value: _ValueType) -> _ValueType: _return_value = self.load(value) return _to_loader(other).load(_return_value) return Loader(_load) def __rrshift__(self, other) -> 'ILoaderClass': """ Overview: Combine two loaders. Arguments: - other (:obj:`ILoaderClass`): The other loader. """ return Loader(other) >> self