ding.utils.loader.collection

ding.utils.loader.collection

CollectionError

Bases: CompositeStructureError

Overview

Collection error.

Interfaces: __init__, errors Properties: errors

errors property

Overview

Get the errors.

__init__(errors)

Overview

Initialize the CollectionError.

Arguments: - errors (:obj:COLLECTION_ERRORS): The errors.

collection(loader, type_back=True)

Overview

Create a collection loader.

Arguments: - loader (:obj:ILoaderClass): The loader. - type_back (:obj:bool): Whether to convert the type back.

tuple_(*loaders)

Overview

Create a tuple loader.

Arguments: - loaders (:obj:tuple): The loaders.

length(min_length=None, max_length=None)

Overview

Create a length loader.

Arguments: - min_length (:obj:int): The minimum length. - max_length (:obj:int): The maximum length.

length_is(length_)

Overview

Create a length loader.

Arguments: - length_ (:obj:int): The length.

contains(content)

Overview

Create a contains loader.

Arguments: - content (:obj:Any): The content.

cofilter(checker, type_back=True)

Overview

Create a cofilter loader.

Arguments: - checker (:obj:Callable[[Any], bool]): The checker. - type_back (:obj:bool): Whether to convert the type back.

tpselector(*indices)

Overview

Create a tuple selector loader.

Arguments: - indices (:obj:tuple): The indices.

Full Source Code

../ding/utils/loader/collection.py

1from typing import Optional, List, Tuple, Callable, Any 2 3from .base import ILoaderClass, Loader, CAPTURE_EXCEPTIONS 4from .exception import CompositeStructureError 5from .types import method 6 7COLLECTION_ERROR_ITEM = Tuple[int, Exception] 8COLLECTION_ERRORS = List[COLLECTION_ERROR_ITEM] 9 10 11class CollectionError(CompositeStructureError): 12 """ 13 Overview: 14 Collection error. 15 Interfaces: 16 ``__init__``, ``errors`` 17 Properties: 18 ``errors`` 19 """ 20 21 def __init__(self, errors: COLLECTION_ERRORS): 22 """ 23 Overview: 24 Initialize the CollectionError. 25 Arguments: 26 - errors (:obj:`COLLECTION_ERRORS`): The errors. 27 """ 28 29 self.__errors = list(errors or []) 30 CompositeStructureError.__init__( 31 self, '{count} error(s) found in collection.'.format(count=repr(list(self.__errors))) 32 ) 33 34 @property 35 def errors(self) -> COLLECTION_ERRORS: 36 """ 37 Overview: 38 Get the errors. 39 """ 40 41 return self.__errors 42 43 44def collection(loader, type_back: bool = True) -> ILoaderClass: 45 """ 46 Overview: 47 Create a collection loader. 48 Arguments: 49 - loader (:obj:`ILoaderClass`): The loader. 50 - type_back (:obj:`bool`): Whether to convert the type back. 51 """ 52 53 loader = Loader(loader) 54 55 def _load(value): 56 _result = [] 57 _errors = [] 58 59 for index, item in enumerate(value): 60 try: 61 _return = loader.load(item) 62 except CAPTURE_EXCEPTIONS as err: 63 _errors.append((index, err)) 64 else: 65 _result.append(_return) 66 67 if _errors: 68 raise CollectionError(_errors) 69 70 if type_back: 71 _result = type(value)(_result) 72 return _result 73 74 return method('__iter__') & Loader(_load) 75 76 77def tuple_(*loaders) -> ILoaderClass: 78 """ 79 Overview: 80 Create a tuple loader. 81 Arguments: 82 - loaders (:obj:`tuple`): The loaders. 83 """ 84 85 loaders = [Loader(loader) for loader in loaders] 86 87 def _load(value: tuple): 88 return tuple([loader(item) for loader, item in zip(loaders, value)]) 89 90 return tuple & length_is(len(loaders)) & Loader(_load) 91 92 93def length(min_length: Optional[int] = None, max_length: Optional[int] = None) -> ILoaderClass: 94 """ 95 Overview: 96 Create a length loader. 97 Arguments: 98 - min_length (:obj:`int`): The minimum length. 99 - max_length (:obj:`int`): The maximum length. 100 """ 101 102 def _load(value): 103 _length = len(value) 104 if min_length is not None and _length < min_length: 105 raise ValueError( 106 'minimum length is {expect}, but {actual} found'.format(expect=repr(min_length), actual=repr(_length)) 107 ) 108 if max_length is not None and _length > max_length: 109 raise ValueError( 110 'maximum length is {expect}, but {actual} found'.format(expect=repr(max_length), actual=repr(_length)) 111 ) 112 113 return value 114 115 return method('__len__') & Loader(_load) 116 117 118def length_is(length_: int) -> ILoaderClass: 119 """ 120 Overview: 121 Create a length loader. 122 Arguments: 123 - length_ (:obj:`int`): The length. 124 """ 125 126 return length(min_length=length_, max_length=length_) 127 128 129def contains(content) -> ILoaderClass: 130 """ 131 Overview: 132 Create a contains loader. 133 Arguments: 134 - content (:obj:`Any`): The content. 135 """ 136 137 def _load(value): 138 if content not in value: 139 raise ValueError('{content} not found in value'.format(content=repr(content))) 140 141 return value 142 143 return method('__contains__') & Loader(_load) 144 145 146def cofilter(checker: Callable[[Any], bool], type_back: bool = True) -> ILoaderClass: 147 """ 148 Overview: 149 Create a cofilter loader. 150 Arguments: 151 - checker (:obj:`Callable[[Any], bool]`): The checker. 152 - type_back (:obj:`bool`): Whether to convert the type back. 153 """ 154 155 def _load(value): 156 _result = [item for item in value if checker(item)] 157 if type_back: 158 _result = type(value)(_result) 159 return _result 160 161 return method('__iter__') & Loader(_load) 162 163 164def tpselector(*indices) -> ILoaderClass: 165 """ 166 Overview: 167 Create a tuple selector loader. 168 Arguments: 169 - indices (:obj:`tuple`): The indices. 170 """ 171 172 def _load(value: tuple): 173 return tuple([value[index] for index in indices]) 174 175 return tuple & Loader(_load)