Shortcuts

Source code for ding.utils.loader.collection

from typing import Optional, List, Tuple, Callable, Any

from .base import ILoaderClass, Loader, CAPTURE_EXCEPTIONS
from .exception import CompositeStructureError
from .types import method

COLLECTION_ERROR_ITEM = Tuple[int, Exception]
COLLECTION_ERRORS = List[COLLECTION_ERROR_ITEM]


[docs]class CollectionError(CompositeStructureError): """ Overview: Collection error. Interfaces: ``__init__``, ``errors`` Properties: ``errors`` """
[docs] def __init__(self, errors: COLLECTION_ERRORS): """ Overview: Initialize the CollectionError. Arguments: - errors (:obj:`COLLECTION_ERRORS`): The errors. """ self.__errors = list(errors or []) CompositeStructureError.__init__( self, '{count} error(s) found in collection.'.format(count=repr(list(self.__errors))) )
@property def errors(self) -> COLLECTION_ERRORS: """ Overview: Get the errors. """ return self.__errors
[docs]def collection(loader, type_back: bool = True) -> ILoaderClass: """ Overview: Create a collection loader. Arguments: - loader (:obj:`ILoaderClass`): The loader. - type_back (:obj:`bool`): Whether to convert the type back. """ loader = Loader(loader) def _load(value): _result = [] _errors = [] for index, item in enumerate(value): try: _return = loader.load(item) except CAPTURE_EXCEPTIONS as err: _errors.append((index, err)) else: _result.append(_return) if _errors: raise CollectionError(_errors) if type_back: _result = type(value)(_result) return _result return method('__iter__') & Loader(_load)
[docs]def tuple_(*loaders) -> ILoaderClass: """ Overview: Create a tuple loader. Arguments: - loaders (:obj:`tuple`): The loaders. """ loaders = [Loader(loader) for loader in loaders] def _load(value: tuple): return tuple([loader(item) for loader, item in zip(loaders, value)]) return tuple & length_is(len(loaders)) & Loader(_load)
[docs]def length(min_length: Optional[int] = None, max_length: Optional[int] = None) -> ILoaderClass: """ Overview: Create a length loader. Arguments: - min_length (:obj:`int`): The minimum length. - max_length (:obj:`int`): The maximum length. """ def _load(value): _length = len(value) if min_length is not None and _length < min_length: raise ValueError( 'minimum length is {expect}, but {actual} found'.format(expect=repr(min_length), actual=repr(_length)) ) if max_length is not None and _length > max_length: raise ValueError( 'maximum length is {expect}, but {actual} found'.format(expect=repr(max_length), actual=repr(_length)) ) return value return method('__len__') & Loader(_load)
[docs]def length_is(length_: int) -> ILoaderClass: """ Overview: Create a length loader. Arguments: - length_ (:obj:`int`): The length. """ return length(min_length=length_, max_length=length_)
[docs]def contains(content) -> ILoaderClass: """ Overview: Create a contains loader. Arguments: - content (:obj:`Any`): The content. """ def _load(value): if content not in value: raise ValueError('{content} not found in value'.format(content=repr(content))) return value return method('__contains__') & Loader(_load)
[docs]def cofilter(checker: Callable[[Any], bool], type_back: bool = True) -> ILoaderClass: """ Overview: Create a cofilter loader. Arguments: - checker (:obj:`Callable[[Any], bool]`): The checker. - type_back (:obj:`bool`): Whether to convert the type back. """ def _load(value): _result = [item for item in value if checker(item)] if type_back: _result = type(value)(_result) return _result return method('__iter__') & Loader(_load)
[docs]def tpselector(*indices) -> ILoaderClass: """ Overview: Create a tuple selector loader. Arguments: - indices (:obj:`tuple`): The indices. """ def _load(value: tuple): return tuple([value[index] for index in indices]) return tuple & Loader(_load)