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# pylint: disable=line-too-long 

22''' Accretive dictionaries. 

23 

24Dictionaries which can grow but never shrink. Once an entry is added, it cannot 

25be modified or removed. 

26 

27* :py:class:`AbstractDictionary`: 

28 Base class defining the accretive dictionary interface. Implementations must 

29 provide ``__getitem__``, ``__iter__``, ``__len__``, and storage methods. 

30 

31* :py:class:`Dictionary`: 

32 Standard implementation of an accretive dictionary. Supports all usual dict 

33 operations except those that would modify or remove existing entries. 

34 

35* :py:class:`ProducerDictionary`: 

36 Automatically generates values for missing keys using a supplied factory 

37 function. Similar to :py:class:`collections.defaultdict` but with accretive 

38 behavior. 

39 

40* :py:class:`ValidatorDictionary`: 

41 Validates entries before addition using a supplied predicate function. 

42 

43* :py:class:`ProducerValidatorDictionary`: 

44 Combines producer and validator behaviors. Generated values must pass 

45 validation before being added. 

46 

47>>> from accretive import Dictionary 

48>>> d = Dictionary( apples = 12, bananas = 6 ) 

49>>> d[ 'cherries' ] = 42 # Add new entry 

50>>> d[ 'apples' ] = 14 # Attempt modification 

51Traceback (most recent call last): 

52 ... 

53accretive.exceptions.EntryImmutabilityError: Cannot alter or remove existing entry for 'apples'. 

54>>> del d[ 'bananas' ] # Attempt removal 

55Traceback (most recent call last): 

56 ... 

57accretive.exceptions.EntryImmutabilityError: Cannot alter or remove existing entry for 'bananas'. 

58 

59>>> from accretive import ProducerDictionary 

60>>> d = ProducerDictionary( list ) # list() called for missing keys 

61>>> d[ 'new' ] 

62[] 

63>>> d[ 'new' ].append( 1 ) # List is mutable, but entry is fixed 

64>>> d[ 'new' ] = [ ] # Attempt modification 

65Traceback (most recent call last): 

66 ... 

67accretive.exceptions.EntryImmutabilityError: Cannot alter or remove existing entry for 'new'. 

68 

69>>> from accretive import ValidatorDictionary 

70>>> d = ValidatorDictionary( lambda k, v: isinstance( v, int ) ) 

71>>> d[ 'valid' ] = 42 # Passes validation 

72>>> d[ 'invalid' ] = 'str' # Fails validation 

73Traceback (most recent call last): 

74 ... 

75accretive.exceptions.EntryValidityError: Cannot add invalid entry with key, 'invalid', and value, 'str', to dictionary. 

76''' 

77# pylint: enable=line-too-long 

78 

79 

80from . import __ 

81from . import classes as _classes 

82from . import objects as _objects 

83 

84 

85class AbstractDictionary( __.cabc.Mapping[ __.H, __.V ] ): 

86 ''' Abstract base class for dictionaries that can grow but not shrink. 

87 

88 An accretive dictionary allows new entries to be added but prevents 

89 modification or removal of existing entries. This provides a middle 

90 ground between immutable and fully mutable mappings. 

91 

92 Implementations must provide: 

93 - __getitem__, __iter__, __len__ 

94 - _pre_setitem_ for entry validation/preparation 

95 - _store_item_ for storage implementation 

96 ''' 

97 

98 @__.abstract_member_function 

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

100 raise NotImplementedError # pragma: no coverage 

101 

102 @__.abstract_member_function 

103 def __len__( self ) -> int: 

104 raise NotImplementedError # pragma: no coverage 

105 

106 @__.abstract_member_function 

107 def __getitem__( self, key: __.H ) -> __.V: 

108 raise NotImplementedError # pragma: no coverage 

109 

110 def _pre_setitem_( # pylint: disable=no-self-use 

111 self, key: __.H, value: __.V 

112 ) -> tuple[ __.H, __.V ]: 

113 ''' Validates and/or prepares entry before addition. 

114 

115 Should raise appropriate exception if entry is invalid. 

116 ''' 

117 return key, value 

118 

119 @__.abstract_member_function 

120 def _store_item_( self, key: __.H, value: __.V ) -> None: 

121 ''' Stores entry in underlying storage. ''' 

122 raise NotImplementedError # pragma: no coverage 

123 

124 def __setitem__( self, key: __.H, value: __.V ) -> None: 

125 key, value = self._pre_setitem_( key, value ) 

126 if key in self: 

127 from .exceptions import EntryImmutabilityError 

128 raise EntryImmutabilityError( key ) 

129 self._store_item_( key, value ) 

130 

131 def __delitem__( self, key: __.H ) -> None: 

132 from .exceptions import EntryImmutabilityError 

133 raise EntryImmutabilityError( key ) 

134 

135 def setdefault( self, key: __.H, default: __.V ) -> __.V: 

136 ''' Returns value for key, setting it to default if missing. ''' 

137 try: return self[ key ] 

138 except KeyError: 

139 self[ key ] = default 

140 return default 

141 

142 def update( 

143 self, 

144 *iterables: __.DictionaryPositionalArgument[ __.H, __.V ], 

145 **entries: __.DictionaryNominativeArgument[ __.V ], 

146 ) -> __.a.Self: 

147 ''' Adds new entries as a batch. Returns self. ''' 

148 from itertools import chain 

149 updates: list[ tuple[ __.H, __.V ] ] = [ ] 

150 for indicator, value in chain.from_iterable( map( # type: ignore 

151 lambda element: ( # type: ignore 

152 element.items( ) 

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

154 else element 

155 ), 

156 ( *iterables, entries ) 

157 ) ): 

158 indicator_, value_ = ( 

159 self._pre_setitem_( indicator, value ) ) # type: ignore 

160 if indicator_ in self: 

161 from .exceptions import EntryImmutabilityError 

162 raise EntryImmutabilityError( indicator_ ) 

163 updates.append( ( indicator_, value_ ) ) 

164 for indicator, value in updates: self._store_item_( indicator, value ) 

165 return self 

166 

167 

168class _DictionaryOperations( AbstractDictionary[ __.H, __.V ] ): 

169 ''' Mix-in providing additional dictionary operations. ''' 

170 

171 def __init__( self, *posargs: __.a.Any, **nomargs: __.a.Any ) -> None: 

172 super( ).__init__( *posargs, **nomargs ) 

173 

174 def __or__( self, other: __.cabc.Mapping[ __.H, __.V ] ) -> __.a.Self: 

175 if not isinstance( other, __.cabc.Mapping ): return NotImplemented 

176 result = self.copy( ) 

177 result.update( other ) 

178 return result 

179 

180 def __ror__( self, other: __.cabc.Mapping[ __.H, __.V ] ) -> __.a.Self: 

181 if not isinstance( other, __.cabc.Mapping ): return NotImplemented 

182 return self | other 

183 

184 def __and__( 

185 self, 

186 other: __.cabc.Set[ __.H ] | __.cabc.Mapping[ __.H, __.V ] 

187 ) -> __.a.Self: 

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

189 return self.with_data( 

190 ( key, value ) for key, value in self.items( ) 

191 if key in other and other[ key ] == value ) 

192 if isinstance( other, ( __.cabc.Set, __.cabc.KeysView ) ): 

193 return self.with_data( 

194 ( key, self[ key ] ) for key in self.keys( ) & other ) 

195 return NotImplemented 

196 

197 def __rand__( 

198 self, 

199 other: __.cabc.Set[ __.H ] | __.cabc.Mapping[ __.H, __.V ] 

200 ) -> __.a.Self: 

201 if not isinstance( 

202 other, ( __.cabc.Mapping, __.cabc.Set, __.cabc.KeysView ) 

203 ): return NotImplemented 

204 return self & other 

205 

206 @__.abstract_member_function 

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

208 ''' Provides fresh copy of dictionary. ''' 

209 raise NotImplementedError # pragma: no coverage 

210 

211 @__.abstract_member_function 

212 def with_data( 

213 self, 

214 *iterables: __.DictionaryPositionalArgument[ __.H, __.V ], 

215 **entries: __.DictionaryNominativeArgument[ __.V ], 

216 ) -> __.a.Self: 

217 ''' Creates new dictionary with same behavior but different data. ''' 

218 raise NotImplementedError # pragma: no coverage 

219 

220 

221 

222class _Dictionary( 

223 __.CoreDictionary[ __.H, __.V ], metaclass = _classes.Class 

224): pass 

225 

226 

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

228 _objects.Object, _DictionaryOperations[ __.H, __.V ] 

229): 

230 ''' Accretive dictionary. ''' 

231 # TODO: version 3.0: Do not subclass from 'Object'. 

232 

233 __slots__ = ( '_data_', ) 

234 

235 _data_: _Dictionary[ __.H, __.V ] 

236 

237 def __init__( 

238 self, 

239 *iterables: __.DictionaryPositionalArgument[ __.H, __.V ], 

240 **entries: __.DictionaryNominativeArgument[ __.V ], 

241 ) -> None: 

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

243 super( ).__init__( ) 

244 

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

246 return iter( self._data_ ) 

247 

248 def __len__( self ) -> int: 

249 return len( self._data_ ) 

250 

251 def __repr__( self ) -> str: 

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

253 fqname = __.calculate_fqname( self ), 

254 contents = str( self._data_ ) ) 

255 

256 def __str__( self ) -> str: 

257 return str( self._data_ ) 

258 

259 def __contains__( self, key: __.a.Any ) -> bool: 

260 return key in self._data_ 

261 

262 def __getitem__( self, key: __.H ) -> __.V: 

263 return self._data_[ key ] 

264 

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

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

267 return self._data_ == other 

268 return NotImplemented 

269 

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

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

272 return self._data_ != other 

273 return NotImplemented 

274 

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

276 ''' Provides fresh copy of dictionary. ''' 

277 return type( self )( self ) 

278 

279 def get( 

280 self, key: __.H, default: __.Optional[ __.V ] = __.absent 

281 ) -> __.a.Annotation[ 

282 __.V, 

283 __.a.Doc( 

284 'Value of entry, if it exists. ' 

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

286 ]: 

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

288 if __.is_absent( default ): 

289 return self._data_.get( key ) # type: ignore 

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

291 

292 def keys( self ) -> __.cabc.KeysView[ __.H ]: 

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

294 return self._data_.keys( ) 

295 

296 def items( self ) -> __.cabc.ItemsView[ __.H, __.V ]: 

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

298 return self._data_.items( ) 

299 

300 def values( self ) -> __.cabc.ValuesView[ __.V ]: 

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

302 return self._data_.values( ) 

303 

304 def with_data( 

305 self, 

306 *iterables: __.DictionaryPositionalArgument[ __.H, __.V ], 

307 **entries: __.DictionaryNominativeArgument[ __.V ], 

308 ) -> __.a.Self: 

309 return type( self )( *iterables, **entries ) 

310 

311 def _store_item_( self, key: __.H, value: __.V ) -> None: 

312 self._data_[ key ] = value 

313 

314Dictionary.__doc__ = __.generate_docstring( 

315 Dictionary, 'dictionary entries accretion' ) 

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

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

318# mixin would provide. 

319__.cabc.Mapping.register( Dictionary ) # type: ignore 

320 

321 

322class ProducerDictionary( Dictionary[ __.H, __.V ] ): 

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

324 

325 __slots__ = ( '_producer_', ) 

326 

327 _producer_: __.DictionaryProducer[ __.V ] 

328 

329 def __init__( 

330 self, 

331 producer: __.DictionaryProducer[ __.V ], 

332 /, 

333 *iterables: __.DictionaryPositionalArgument[ __.H, __.V ], 

334 **entries: __.DictionaryNominativeArgument[ __.V ], 

335 ): 

336 # TODO: Validate producer argument. 

337 self._producer_ = producer 

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

339 

340 def __repr__( self ) -> str: 

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

342 fqname = __.calculate_fqname( self ), 

343 producer = self._producer_, 

344 contents = str( self._data_ ) ) 

345 

346 def __getitem__( self, key: __.H ) -> __.V: 

347 if key not in self: 

348 value = self._producer_( ) 

349 self[ key ] = value 

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

351 return value 

352 

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

354 ''' Provides fresh copy of dictionary. ''' 

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

356 return dictionary.update( self ) 

357 

358 def setdefault( self, key: __.H, default: __.V ) -> __.V: 

359 ''' Returns value for key, setting it to default if missing. ''' 

360 if key not in self: 

361 self[ key ] = default 

362 return default 

363 return self[ key ] 

364 

365 def with_data( 

366 self, 

367 *iterables: __.DictionaryPositionalArgument[ __.H, __.V ], 

368 **entries: __.DictionaryNominativeArgument[ __.V ], 

369 ) -> __.a.Self: 

370 return type( self )( self._producer_, *iterables, **entries ) 

371 

372ProducerDictionary.__doc__ = __.generate_docstring( 

373 ProducerDictionary, 

374 'dictionary entries accretion', 

375 'dictionary entries production', 

376) 

377 

378 

379class ValidatorDictionary( Dictionary[ __.H, __.V ] ): 

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

381 

382 __slots__ = ( '_validator_', ) 

383 

384 _validator_: __.DictionaryValidator[ __.H, __.V ] 

385 

386 def __init__( 

387 self, 

388 validator: __.DictionaryValidator[ __.H, __.V ], 

389 /, 

390 *iterables: __.DictionaryPositionalArgument[ __.H, __.V ], 

391 **entries: __.DictionaryNominativeArgument[ __.V ], 

392 ) -> None: 

393 self._validator_ = validator 

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

395 

396 def __repr__( self ) -> str: 

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

398 fqname = __.calculate_fqname( self ), 

399 validator = self._validator_, 

400 contents = str( self._data_ ) ) 

401 

402 def _pre_setitem_( self, key: __.H, value: __.V ) -> tuple[ __.H, __.V ]: 

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

404 from .exceptions import EntryValidityError 

405 raise EntryValidityError( key, value ) 

406 return key, value 

407 

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

409 ''' Provides fresh copy of dictionary. ''' 

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

411 return dictionary.update( self ) 

412 

413 def with_data( 

414 self, 

415 *iterables: __.DictionaryPositionalArgument[ __.H, __.V ], 

416 **entries: __.DictionaryNominativeArgument[ __.V ], 

417 ) -> __.a.Self: 

418 return type( self )( self._validator_, *iterables, **entries ) 

419 

420ValidatorDictionary.__doc__ = __.generate_docstring( 

421 ValidatorDictionary, 

422 'dictionary entries accretion', 

423 'dictionary entries validation', 

424) 

425 

426 

427class ProducerValidatorDictionary( Dictionary[ __.H, __.V ] ): 

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

429 

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

431 

432 _producer_: __.DictionaryProducer[ __.V ] 

433 _validator_: __.DictionaryValidator[ __.H, __.V ] 

434 

435 def __init__( 

436 self, 

437 producer: __.DictionaryProducer[ __.V ], 

438 validator: __.DictionaryValidator[ __.H, __.V ], 

439 /, 

440 *iterables: __.DictionaryPositionalArgument[ __.H, __.V ], 

441 **entries: __.DictionaryNominativeArgument[ __.V ], 

442 ) -> None: 

443 self._producer_ = producer 

444 self._validator_ = validator 

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

446 

447 def __repr__( self ) -> str: 

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

449 fqname = __.calculate_fqname( self ), 

450 producer = self._producer_, 

451 validator = self._validator_, 

452 contents = str( self._data_ ) ) 

453 

454 def __getitem__( self, key: __.H ) -> __.V: 

455 if key not in self: 

456 value = self._producer_( ) 

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

458 from .exceptions import EntryValidityError 

459 raise EntryValidityError( key, value ) 

460 self[ key ] = value 

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

462 return value 

463 

464 def _pre_setitem_( self, key: __.H, value: __.V ) -> tuple[ __.H, __.V ]: 

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

466 from .exceptions import EntryValidityError 

467 raise EntryValidityError( key, value ) 

468 return key, value 

469 

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

471 ''' Provides fresh copy of dictionary. ''' 

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

473 return dictionary.update( self ) 

474 

475 def setdefault( self, key: __.H, default: __.V ) -> __.V: 

476 ''' Returns value for key, setting it to default if missing. ''' 

477 if key not in self: 

478 self[ key ] = default 

479 return default 

480 return self[ key ] 

481 

482 def with_data( 

483 self, 

484 *iterables: __.DictionaryPositionalArgument[ __.H, __.V ], 

485 **entries: __.DictionaryNominativeArgument[ __.V ], 

486 ) -> __.a.Self: 

487 return type( self )( 

488 self._producer_, self._validator_, *iterables, **entries ) 

489 

490ProducerValidatorDictionary.__doc__ = __.generate_docstring( 

491 ProducerValidatorDictionary, 

492 'dictionary entries accretion', 

493 'dictionary entries production', 

494 'dictionary entries validation', 

495)