Shortcuts

Source code for ding.utils.loader.base

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)


[docs]class ILoaderClass: """ Overview: Base class of loader. Interfaces: ``__init__``, ``_load``, ``load``, ``check``, ``__call__``, ``__and__``, ``__or__``, ``__rshift__`` """
[docs] @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
[docs] def load(self, value: _ValueType) -> _ValueType: """ Overview: Load the value. Arguments: - value (:obj:`_ValueType`): The value to be loaded. """ return self.__load(value)
[docs] 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