Coverage for sources/accretive/dictionaries.py: 100%

1# vim: set filetype=python fileencoding=utf-8: 

2# -*- coding: utf-8 -*- 

3 

4#============================================================================# 

5# # 

6# Licensed under the Apache License, Version 2.0 (the "License"); # 

7# you may not use this file except in compliance with the License. # 

8# You may obtain a copy of the License at # 

9# # 

10# http://www.apache.org/licenses/LICENSE-2.0 # 

11# # 

12# Unless required by applicable law or agreed to in writing, software # 

13# distributed under the License is distributed on an "AS IS" BASIS, # 

14# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # 

15# See the License for the specific language governing permissions and # 

16# limitations under the License. # 

17# # 

18#============================================================================# 

19 

20 

21''' Accretive dictionaries. ''' 

22 

23 

24from . import __ 

25from . import classes as _classes 

26from . import objects as _objects 

27 

28 

29class _Dictionary( # type: ignore 

30 __.CoreDictionary, metaclass = _classes.Class 

31): pass 

32 

33 

34class Dictionary( # pylint: disable=eq-without-hash 

35 _objects.Object, 

36 __.a.Generic[ __.H, __.V ], # type: ignore[misc] 

37): 

38 ''' Accretive dictionary. ''' 

39 

40 __slots__ = ( '_data_', ) 

41 

42 _data_: _Dictionary 

43 

44 def __init__( 

45 self, 

46 *iterables: __.DictionaryPositionalArgument, 

47 **entries: __.DictionaryNominativeArgument, 

48 ) -> None: 

49 self._data_ = _Dictionary( *iterables, **entries ) 

50 super( ).__init__( ) 

51 

52 def __iter__( self ) -> __.cabc.Iterator[ __.H ]: 

53 return iter( self._data_ ) 

54 

55 def __len__( self ) -> int: 

56 return len( self._data_ ) 

57 

58 def __repr__( self ) -> str: 

59 return "{fqname}( {contents} )".format( 

60 fqname = __.calculate_fqname( self ), 

61 contents = str( self._data_ ) ) 

62 

63 def __str__( self ) -> str: 

64 return str( self._data_ ) 

65 

66 def __contains__( self, key: __.cabc.Hashable ) -> bool: 

67 return key in self._data_ 

68 

69 def __delitem__( self, key: __.cabc.Hashable ) -> None: 

70 from .exceptions import IndelibleEntryError 

71 raise IndelibleEntryError( key ) 

72 

73 def __getitem__( self, key: __.cabc.Hashable ) -> __.a.Any: 

74 return self._data_[ key ] 

75 

76 def __setitem__( self, key: __.cabc.Hashable, value: __.a.Any ) -> None: 

77 self._data_[ key ] = value 

78 

79 def __eq__( self, other: __.a.Any ) -> __.ComparisonResult: 

80 if isinstance( other, __.cabc.Mapping ): 

81 return self._data_ == other 

82 return NotImplemented 

83 

84 def __ne__( self, other: __.a.Any ) -> __.ComparisonResult: 

85 if isinstance( other, __.cabc.Mapping ): 

86 return self._data_ != other 

87 return NotImplemented 

88 

89 def copy( self ) -> __.a.Self: 

90 ''' Provides fresh copy of dictionary. ''' 

91 return type( self )( self ) 

92 

93 def get( 

94 self, 

95 key: __.cabc.Hashable, 

96 default: __.Optional[ __.a.Any ] = __.absent, 

97 ) -> __.a.Annotation[ 

98 __.a.Any, 

99 __.a.Doc( 

100 'Value of entry, if it exists. ' 

101 'Else, supplied default value or ``None``.' ) 

102 ]: 

103 ''' Retrieves entry associated with key, if it exists. ''' 

104 if __.is_absent( default ): return self._data_.get( key ) 

105 return self._data_.get( key, default ) 

106 

107 def update( 

108 self, 

109 *iterables: __.DictionaryPositionalArgument, 

110 **entries: __.DictionaryNominativeArgument, 

111 ) -> __.a.Self: 

112 ''' Adds new entries as a batch. ''' 

113 self._data_.update( *iterables, **entries ) 

114 return self 

115 

116 def keys( self ) -> __.cabc.KeysView[ __.cabc.Hashable ]: 

117 ''' Provides iterable view over dictionary keys. ''' 

118 return self._data_.keys( ) 

119 

120 def items( self ) -> __.cabc.ItemsView[ __.cabc.Hashable, __.a.Any ]: 

121 ''' Provides iterable view over dictionary items. ''' 

122 return self._data_.items( ) 

123 

124 def values( self ) -> __.cabc.ValuesView[ __.a.Any ]: 

125 ''' Provides iterable view over dictionary values. ''' 

126 return self._data_.values( ) 

127 

128Dictionary.__doc__ = __.generate_docstring( 

129 Dictionary, 

130 'dictionary entries accretion', 

131 'instance attributes accretion', 

132) 

133# Register as subclass of Mapping rather than use it as mixin. 

134# We directly implement, for the sake of efficiency, the methods which the 

135# mixin would provide. 

136__.cabc.Mapping.register( Dictionary ) 

137 

138 

139class ProducerDictionary( 

140 Dictionary, 

141 __.a.Generic[ __.H, __.V ], # type: ignore[misc] 

142): 

143 ''' Accretive dictionary with default value for missing entries. ''' 

144 

145 __slots__ = ( '_producer_', ) 

146 

147 _producer_: __.DictionaryProducer 

148 

149 def __init__( 

150 self, 

151 producer: __.DictionaryProducer, 

152 /, 

153 *iterables: __.DictionaryPositionalArgument, 

154 **entries: __.DictionaryNominativeArgument 

155 ): 

156 # TODO: Validate producer argument. 

157 self._producer_ = producer 

158 super( ).__init__( *iterables, **entries ) 

159 

160 def __repr__( self ) -> str: 

161 return "{fqname}( {producer}, {contents} )".format( 

162 fqname = __.calculate_fqname( self ), 

163 producer = self._producer_, 

164 contents = str( self._data_ ) ) 

165 

166 def __getitem__( self, key: __.cabc.Hashable ) -> __.a.Any: 

167 if key not in self: 

168 value = self._producer_( ) 

169 self[ key ] = value 

170 else: value = super( ).__getitem__( key ) 

171 return value 

172 

173 def copy( self ) -> __.a.Self: 

174 ''' Provides fresh copy of dictionary. ''' 

175 dictionary = type( self )( self._producer_ ) 

176 return dictionary.update( self ) 

177 

178ProducerDictionary.__doc__ = __.generate_docstring( 

179 ProducerDictionary, 

180 'dictionary entries accretion', 

181 'dictionary entries production', 

182 'instance attributes accretion', 

183) 

184 

185 

186class ValidatorDictionary( 

187 Dictionary, 

188 __.a.Generic[ __.H, __.V ], # type: ignore[misc] 

189): 

190 ''' Accretive dictionary with validation of new entries. ''' 

191 

192 __slots__ = ( '_validator_', ) 

193 

194 _validator_: __.DictionaryValidator 

195 

196 def __init__( 

197 self, 

198 validator: __.DictionaryValidator, 

199 /, 

200 *iterables: __.DictionaryPositionalArgument, 

201 **entries: __.DictionaryNominativeArgument, 

202 ) -> None: 

203 self._validator_ = validator 

204 super( ).__init__( *iterables, **entries ) 

205 

206 def __repr__( self ) -> str: 

207 return "{fqname}( {validator}, {contents} )".format( 

208 fqname = __.calculate_fqname( self ), 

209 validator = self._validator_, 

210 contents = str( self._data_ ) ) 

211 

212 def __setitem__( self, key: __.cabc.Hashable, value: __.a.Any ) -> None: 

213 if not self._validator_( key, value ): 

214 from .exceptions import EntryValidationError 

215 raise EntryValidationError( key, value ) 

216 super( ).__setitem__( key, value ) 

217 

218 def copy( self ) -> __.a.Self: 

219 ''' Provides fresh copy of dictionary. ''' 

220 dictionary = type( self )( self._validator_ ) 

221 return dictionary.update( self ) 

222 

223 def update( 

224 self, 

225 *iterables: __.DictionaryPositionalArgument, 

226 **entries: __.DictionaryNominativeArgument, 

227 ) -> __.a.Self: 

228 ''' Adds new entries as a batch. ''' 

229 from itertools import chain 

230 # Validate all entries before adding any 

231 for indicator, value in chain.from_iterable( map( 

232 lambda element: ( 

233 element.items( ) 

234 if isinstance( element, __.cabc.Mapping ) 

235 else element 

236 ), 

237 ( *iterables, entries ) 

238 ) ): 

239 if not self._validator_( indicator, value ): 

240 from .exceptions import EntryValidationError 

241 raise EntryValidationError( indicator, value ) 

242 return super( ).update( *iterables, **entries ) 

243 

244ValidatorDictionary.__doc__ = __.generate_docstring( 

245 ValidatorDictionary, 

246 'dictionary entries accretion', 

247 'dictionary entries validation', 

248 'instance attributes accretion', 

249) 

250 

251 

252class ProducerValidatorDictionary( 

253 Dictionary, 

254 __.a.Generic[ __.H, __.V ], # type: ignore[misc] 

255): 

256 ''' Accretive dictionary with defaults and validation. ''' 

257 

258 __slots__ = ( '_producer_', '_validator_' ) 

259 

260 _producer_: __.DictionaryProducer 

261 _validator_: __.DictionaryValidator 

262 

263 def __init__( 

264 self, 

265 producer: __.DictionaryProducer, 

266 validator: __.DictionaryValidator, 

267 /, 

268 *iterables: __.DictionaryPositionalArgument, 

269 **entries: __.DictionaryNominativeArgument, 

270 ) -> None: 

271 self._producer_ = producer 

272 self._validator_ = validator 

273 super( ).__init__( *iterables, **entries ) 

274 

275 def __repr__( self ) -> str: 

276 return "{fqname}( {producer}, {validator}, {contents} )".format( 

277 fqname = __.calculate_fqname( self ), 

278 producer = self._producer_, 

279 validator = self._validator_, 

280 contents = str( self._data_ ) ) 

281 

282 def __getitem__( self, key: __.cabc.Hashable ) -> __.a.Any: 

283 if key not in self: 

284 value = self._producer_( ) 

285 if not self._validator_( key, value ): 

286 from .exceptions import EntryValidationError 

287 raise EntryValidationError( key, value ) 

288 self[ key ] = value 

289 else: value = super( ).__getitem__( key ) 

290 return value 

291 

292 def __setitem__( self, key: __.cabc.Hashable, value: __.a.Any ) -> None: 

293 if not self._validator_( key, value ): 

294 from .exceptions import EntryValidationError 

295 raise EntryValidationError( key, value ) 

296 super( ).__setitem__( key, value ) 

297 

298 def copy( self ) -> __.a.Self: 

299 ''' Provides fresh copy of dictionary. ''' 

300 dictionary = type( self )( self._producer_, self._validator_ ) 

301 return dictionary.update( self ) 

302 

303 def update( 

304 self, 

305 *iterables: __.DictionaryPositionalArgument, 

306 **entries: __.DictionaryNominativeArgument, 

307 ) -> __.a.Self: 

308 ''' Adds new entries as a batch. ''' 

309 from itertools import chain 

310 # Validate all entries before adding any 

311 for indicator, value in chain.from_iterable( map( 

312 lambda element: ( 

313 element.items( ) 

314 if isinstance( element, __.cabc.Mapping ) 

315 else element 

316 ), 

317 ( *iterables, entries ) 

318 ) ): 

319 if not self._validator_( indicator, value ): 

320 from .exceptions import EntryValidationError 

321 raise EntryValidationError( indicator, value ) 

322 return super( ).update( *iterables, **entries ) 

323 

324ProducerValidatorDictionary.__doc__ = __.generate_docstring( 

325 ProducerValidatorDictionary, 

326 'dictionary entries accretion', 

327 'dictionary entries production', 

328 'dictionary entries validation', 

329 'instance attributes accretion', 

330)