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 

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

25 cannot be modified or removed. 

26 

27 * :py:class:`AbstractDictionary`: 

28 Base class defining the accretive dictionary interface. Implementations 

29 must provide ``__getitem__``, ``__iter__``, ``__len__``, and storage 

30 methods. 

31 

32 * :py:class:`Dictionary`: 

33 Standard implementation of an accretive dictionary. Supports all usual 

34 dict operations except those that would modify or remove existing 

35 entries. 

36 

37 * :py:class:`ProducerDictionary`: 

38 Automatically generates values for missing keys using a supplied factory 

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

40 accretive behavior. 

41 

42 * :py:class:`ValidatorDictionary`: 

43 Validates entries before addition using a supplied predicate function. 

44 

45 * :py:class:`ProducerValidatorDictionary`: 

46 Combines producer and validator behaviors. Generated values must pass 

47 validation before being added. 

48 

49 >>> from accretive import Dictionary 

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

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

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

53 Traceback (most recent call last): 

54 ... 

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

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

57 Traceback (most recent call last): 

58 ... 

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

60 

61 >>> from accretive import ProducerDictionary 

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

63 >>> d[ 'new' ] 

64 [] 

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

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

67 Traceback (most recent call last): 

68 ... 

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

70 

71 >>> from accretive import ValidatorDictionary 

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

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

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

75 Traceback (most recent call last): 

76 ... 

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

78''' 

79# pylint: enable=line-too-long 

80 

81 

82from . import __ 

83from . import classes as _classes 

84 

85 

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

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

88 

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

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

91 ground between immutable and fully mutable mappings. 

92 

93 Implementations must provide: 

94 - __getitem__, __iter__, __len__ 

95 - _pre_setitem_ for entry validation/preparation 

96 - _store_item_ for storage implementation 

97 ''' 

98 

99 @__.abc.abstractmethod 

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

101 raise NotImplementedError # pragma: no coverage 

102 

103 @__.abc.abstractmethod 

104 def __len__( self ) -> int: 

105 raise NotImplementedError # pragma: no coverage 

106 

107 @__.abc.abstractmethod 

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

109 raise NotImplementedError # pragma: no coverage 

110 

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

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

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

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

115 

116 Should raise appropriate exception if entry is invalid. 

117 ''' 

118 return key, value 

119 

120 @__.abc.abstractmethod 

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

122 ''' Stores entry in underlying storage. ''' 

123 raise NotImplementedError # pragma: no coverage 

124 

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

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

127 if key in self: 

128 from .exceptions import EntryImmutabilityError 

129 raise EntryImmutabilityError( key ) 

130 self._store_item_( key, value ) 

131 

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

133 from .exceptions import EntryImmutabilityError 

134 raise EntryImmutabilityError( key ) 

135 

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

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

138 try: return self[ key ] 

139 except KeyError: 

140 self[ key ] = default 

141 return default 

142 

143 def update( 

144 self, 

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

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

147 ) -> __.typx.Self: 

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

149 from itertools import chain 

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

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

152 lambda element: ( # type: ignore 

153 element.items( ) 

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

155 else element 

156 ), 

157 ( *iterables, entries ) 

158 ) ): 

159 indicator_, value_ = ( 

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

161 if indicator_ in self: 

162 from .exceptions import EntryImmutabilityError 

163 raise EntryImmutabilityError( indicator_ ) 

164 updates.append( ( indicator_, value_ ) ) 

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

166 return self 

167 

168 

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

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

171 

172 def __init__( 

173 self, *posargs: __.typx.Any, **nomargs: __.typx.Any 

174 ) -> None: 

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

176 

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

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

179 conflicts = set( self.keys( ) ) & set( other.keys( ) ) 

180 if conflicts: 

181 from .exceptions import EntryImmutabilityError 

182 raise EntryImmutabilityError( next( iter( conflicts ) ) ) 

183 data = dict( self ) 

184 data.update( other ) 

185 return self.with_data( data ) 

186 

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

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

189 return self | other 

190 

191 def __and__( 

192 self, 

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

194 ) -> __.typx.Self: 

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

196 return self.with_data( # pyright: ignore 

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

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

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

200 return self.with_data( # pyright: ignore 

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

202 return NotImplemented 

203 

204 def __rand__( 

205 self, 

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

207 ) -> __.typx.Self: 

208 if not isinstance( 

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

210 ): return NotImplemented 

211 return self & other 

212 

213 @__.abc.abstractmethod 

214 def copy( self ) -> __.typx.Self: 

215 ''' Provides fresh copy of dictionary. ''' 

216 raise NotImplementedError # pragma: no coverage 

217 

218 @__.abc.abstractmethod 

219 def with_data( 

220 self, 

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

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

223 ) -> __.typx.Self: 

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

225 raise NotImplementedError # pragma: no coverage 

226 

227 

228class _Dictionary( 

229 __.AccretiveDictionary[ __.H, __.V ], metaclass = _classes.Class 

230): pass 

231 

232 

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

234 _DictionaryOperations[ __.H, __.V ] 

235): 

236 ''' Accretive dictionary. ''' 

237 

238 __slots__ = ( '_data_', ) 

239 

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

241 

242 def __init__( 

243 self, 

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

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

246 ) -> None: 

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

248 super( ).__init__( ) 

249 

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

251 return iter( self._data_ ) 

252 

253 def __len__( self ) -> int: 

254 return len( self._data_ ) 

255 

256 def __repr__( self ) -> str: 

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

258 fqname = __.calculate_fqname( self ), 

259 contents = str( self._data_ ) ) 

260 

261 def __str__( self ) -> str: 

262 return str( self._data_ ) 

263 

264 def __contains__( self, key: __.typx.Any ) -> bool: 

265 return key in self._data_ 

266 

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

268 return self._data_[ key ] 

269 

270 def __eq__( self, other: __.typx.Any ) -> __.ComparisonResult: 

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

272 return self._data_ == other 

273 return NotImplemented 

274 

275 def __ne__( self, other: __.typx.Any ) -> __.ComparisonResult: 

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

277 return self._data_ != other 

278 return NotImplemented 

279 

280 def copy( self ) -> __.typx.Self: 

281 ''' Provides fresh copy of dictionary. ''' 

282 return type( self )( self ) 

283 

284 def get( 

285 self, key: __.H, default: __.Absential[ __.V ] = __.absent 

286 ) -> __.typx.Annotated[ 

287 __.V, 

288 __.typx.Doc( 

289 'Value of entry, if it exists. ' 

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

291 ]: 

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

293 if __.is_absent( default ): 

294 return self._data_.get( key ) # pyright: ignore 

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

296 

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

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

299 return self._data_.keys( ) 

300 

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

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

303 return self._data_.items( ) 

304 

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

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

307 return self._data_.values( ) 

308 

309 def with_data( 

310 self, 

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

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

313 ) -> __.typx.Self: 

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

315 

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

317 self._data_[ key ] = value 

318 

319Dictionary.__doc__ = __.generate_docstring( 

320 Dictionary, 'dictionary entries accretion' ) 

321 

322 

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

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

325 

326 __slots__ = ( '_producer_', ) 

327 

328 _producer_: __.DictionaryProducer[ __.V ] 

329 

330 def __init__( 

331 self, 

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

333 /, 

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

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

336 ): 

337 # TODO: Validate producer argument. 

338 self._producer_ = producer 

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

340 

341 def __repr__( self ) -> str: 

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

343 fqname = __.calculate_fqname( self ), 

344 producer = self._producer_, 

345 contents = str( self._data_ ) ) 

346 

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

348 if key not in self: 

349 value = self._producer_( ) 

350 self[ key ] = value 

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

352 return value 

353 

354 def copy( self ) -> __.typx.Self: 

355 ''' Provides fresh copy of dictionary. ''' 

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

357 return dictionary.update( self ) 

358 

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

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

361 if key not in self: 

362 self[ key ] = default 

363 return default 

364 return self[ key ] 

365 

366 def with_data( 

367 self, 

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

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

370 ) -> __.typx.Self: 

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

372 

373ProducerDictionary.__doc__ = __.generate_docstring( 

374 ProducerDictionary, 

375 'dictionary entries accretion', 

376 'dictionary entries production', 

377) 

378 

379 

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

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

382 

383 __slots__ = ( '_validator_', ) 

384 

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

386 

387 def __init__( 

388 self, 

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

390 /, 

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

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

393 ) -> None: 

394 self._validator_ = validator 

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

396 

397 def __repr__( self ) -> str: 

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

399 fqname = __.calculate_fqname( self ), 

400 validator = self._validator_, 

401 contents = str( self._data_ ) ) 

402 

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

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

405 from .exceptions import EntryValidityError 

406 raise EntryValidityError( key, value ) 

407 return key, value 

408 

409 def copy( self ) -> __.typx.Self: 

410 ''' Provides fresh copy of dictionary. ''' 

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

412 return dictionary.update( self ) 

413 

414 def with_data( 

415 self, 

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

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

418 ) -> __.typx.Self: 

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

420 

421ValidatorDictionary.__doc__ = __.generate_docstring( 

422 ValidatorDictionary, 

423 'dictionary entries accretion', 

424 'dictionary entries validation', 

425) 

426 

427 

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

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

430 

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

432 

433 _producer_: __.DictionaryProducer[ __.V ] 

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

435 

436 def __init__( 

437 self, 

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

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

440 /, 

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

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

443 ) -> None: 

444 self._producer_ = producer 

445 self._validator_ = validator 

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

447 

448 def __repr__( self ) -> str: 

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

450 fqname = __.calculate_fqname( self ), 

451 producer = self._producer_, 

452 validator = self._validator_, 

453 contents = str( self._data_ ) ) 

454 

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

456 if key not in self: 

457 value = self._producer_( ) 

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

459 from .exceptions import EntryValidityError 

460 raise EntryValidityError( key, value ) 

461 self[ key ] = value 

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

463 return value 

464 

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

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

467 from .exceptions import EntryValidityError 

468 raise EntryValidityError( key, value ) 

469 return key, value 

470 

471 def copy( self ) -> __.typx.Self: 

472 ''' Provides fresh copy of dictionary. ''' 

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

474 return dictionary.update( self ) 

475 

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

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

478 if key not in self: 

479 self[ key ] = default 

480 return default 

481 return self[ key ] 

482 

483 def with_data( 

484 self, 

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

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

487 ) -> __.typx.Self: 

488 return type( self )( 

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

490 

491ProducerValidatorDictionary.__doc__ = __.generate_docstring( 

492 ProducerValidatorDictionary, 

493 'dictionary entries accretion', 

494 'dictionary entries production', 

495 'dictionary entries validation', 

496)