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 Dictionaries which can grow but never shrink. Once an entry is added, it 

24 cannot be modified or removed. 

25 

26 * :py:class:`AbstractDictionary`: 

27 Base class defining the accretive dictionary interface. Implementations 

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

29 methods. 

30 

31 * :py:class:`Dictionary`: 

32 Standard implementation of an accretive dictionary. Supports all usual 

33 dict operations except those that would modify or remove existing 

34 entries. 

35 

36 * :py:class:`ProducerDictionary`: 

37 Automatically generates values for missing keys using a supplied factory 

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

39 accretive behavior. 

40 

41 * :py:class:`ValidatorDictionary`: 

42 Validates entries before addition using a supplied predicate function. 

43 

44 * :py:class:`ProducerValidatorDictionary`: 

45 Combines producer and validator behaviors. Generated values must pass 

46 validation before being added. 

47 

48 >>> from accretive import Dictionary 

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

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

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

52 Traceback (most recent call last): 

53 ... 

54 accretive.exceptions.EntryImmutability: Could not alter or remove existing entry for 'apples'. 

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

56 Traceback (most recent call last): 

57 ... 

58 accretive.exceptions.EntryImmutability: Could not alter or remove existing entry for 'bananas'. 

59 

60 >>> from accretive import ProducerDictionary 

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

62 >>> d[ 'new' ] 

63 [] 

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

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

66 Traceback (most recent call last): 

67 ... 

68 accretive.exceptions.EntryImmutability: Could not alter or remove existing entry for 'new'. 

69 

70 >>> from accretive import ValidatorDictionary 

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

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

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

74 Traceback (most recent call last): 

75 ... 

76 accretive.exceptions.EntryInvalidity: Could not add invalid entry with key, 'invalid', and value, 'str', to dictionary. 

77''' # noqa: E501 

78 

79 

80from . import __ 

81from . import classes as _classes 

82 

83 

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

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

86 

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

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

89 ground between immutable and fully mutable mappings. 

90 

91 Implementations must provide: 

92 - __getitem__, __iter__, __len__ 

93 - _pre_setitem_ for entry validation/preparation 

94 - _store_item_ for storage implementation 

95 ''' 

96 

97 @__.abc.abstractmethod 

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

99 raise NotImplementedError # pragma: no coverage 

100 

101 @__.abc.abstractmethod 

102 def __len__( self ) -> int: 

103 raise NotImplementedError # pragma: no coverage 

104 

105 @__.abc.abstractmethod 

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

107 raise NotImplementedError # pragma: no coverage 

108 

109 def _pre_setitem_( 

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

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

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

113 

114 Should raise appropriate exception if entry is invalid. 

115 ''' 

116 return key, value 

117 

118 @__.abc.abstractmethod 

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

120 ''' Stores entry in underlying storage. ''' 

121 raise NotImplementedError # pragma: no coverage 

122 

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

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

125 if key in self: 

126 from .exceptions import EntryImmutability 

127 raise EntryImmutability( key ) 

128 self._store_item_( key, value ) 

129 

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

131 from .exceptions import EntryImmutability 

132 raise EntryImmutability( key ) 

133 

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

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

136 try: return self[ key ] 

137 except KeyError: 

138 self[ key ] = default 

139 return default 

140 

141 def update( 

142 self, 

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

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

145 ) -> __.typx.Self: 

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

147 from itertools import chain 

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

149 for indicator, value in chain.from_iterable( map( # pyright: ignore 

150 lambda element: ( # pyright: ignore 

151 element.items( ) 

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

153 else element 

154 ), 

155 ( *iterables, entries ) 

156 ) ): 

157 indicator_, value_ = ( 

158 self._pre_setitem_( indicator, value ) ) # pyright: ignore 

159 if indicator_ in self: 

160 from .exceptions import EntryImmutability 

161 raise EntryImmutability( indicator_ ) 

162 updates.append( ( indicator_, value_ ) ) 

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

164 return self 

165 

166 

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

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

169 

170 def __init__( 

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

172 ) -> None: 

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

174 

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

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

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

178 if conflicts: 

179 from .exceptions import EntryImmutability 

180 raise EntryImmutability( next( iter( conflicts ) ) ) 

181 data = dict( self ) 

182 data.update( other ) 

183 return self.with_data( data ) 

184 

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

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

187 return self | other 

188 

189 def __and__( 

190 self, 

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

192 ) -> __.typx.Self: 

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

194 return self.with_data( # pyright: ignore 

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

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

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

198 return self.with_data( # pyright: ignore 

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

200 return NotImplemented 

201 

202 def __rand__( 

203 self, 

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

205 ) -> __.typx.Self: 

206 if not isinstance( 

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

208 ): return NotImplemented 

209 return self & other 

210 

211 @__.abc.abstractmethod 

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

213 ''' Provides fresh copy of dictionary. ''' 

214 raise NotImplementedError # pragma: no coverage 

215 

216 @__.abc.abstractmethod 

217 def with_data( 

218 self, 

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

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

221 ) -> __.typx.Self: 

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

223 raise NotImplementedError # pragma: no coverage 

224 

225 

226class Dictionary( 

227 _DictionaryOperations[ __.H, __.V ], 

228 metaclass = _classes.AbstractBaseClass, 

229): 

230 ''' Accretive dictionary. ''' 

231 

232 __slots__ = ( '_data_', ) 

233 

234 _data_: __.AccretiveDictionary[ __.H, __.V ] 

235 _dynadoc_fragments_ = ( 'dictionary entries accrete', ) 

236 

237 def __init__( 

238 self, 

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

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

241 ) -> None: 

242 self._data_ = __.AccretiveDictionary( *iterables, **entries ) 

243 super( ).__init__( ) 

244 

245 __hash__ = None 

246 

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

248 return iter( self._data_ ) 

249 

250 def __len__( self ) -> int: 

251 return len( self._data_ ) 

252 

253 def __repr__( self ) -> str: 

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

255 fqname = __.ccutils.qualify_class_name( type( self ) ), 

256 contents = str( self._data_ ) ) 

257 

258 def __str__( self ) -> str: 

259 return str( self._data_ ) 

260 

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

262 return key in self._data_ 

263 

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

265 return self._data_[ key ] 

266 

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

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

269 return self._data_ == other 

270 return NotImplemented 

271 

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

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

274 return self._data_ != other 

275 return NotImplemented 

276 

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

278 ''' Provides fresh copy of dictionary. ''' 

279 return type( self )( self ) 

280 

281 def get( # pyright: ignore 

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

283 ) -> __.typx.Annotated[ 

284 __.V, 

285 __.typx.Doc( 

286 'Value of entry, if it exists. ' 

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

288 ]: 

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

290 if __.is_absent( default ): 

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

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

293 

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

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

296 return self._data_.keys( ) 

297 

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

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

300 return self._data_.items( ) 

301 

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

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

304 return self._data_.values( ) 

305 

306 def with_data( 

307 self, 

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

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

310 ) -> __.typx.Self: 

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

312 

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

314 self._data_[ key ] = value 

315 

316 

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

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

319 

320 __slots__ = ( '_producer_', ) 

321 

322 _dynadoc_fragments_ = ( 

323 'dictionary entries accrete', 'dictionary entries produce' ) 

324 _producer_: __.DictionaryProducer[ __.V ] 

325 

326 def __init__( 

327 self, 

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

329 /, 

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

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

332 ): 

333 # TODO: Validate producer argument. 

334 self._producer_ = producer 

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

336 

337 def __repr__( self ) -> str: 

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

339 fqname = __.ccutils.qualify_class_name( type( self ) ), 

340 producer = self._producer_, 

341 contents = str( self._data_ ) ) 

342 

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

344 if key not in self: 

345 value = self._producer_( ) 

346 self[ key ] = value 

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

348 return value 

349 

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

351 ''' Provides fresh copy of dictionary. ''' 

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

353 return dictionary.update( self ) 

354 

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

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

357 if key not in self: 

358 self[ key ] = default 

359 return default 

360 return self[ key ] 

361 

362 def with_data( 

363 self, 

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

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

366 ) -> __.typx.Self: 

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

368 

369 

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

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

372 

373 __slots__ = ( '_validator_', ) 

374 

375 _dynadoc_fragments_ = ( 

376 'dictionary entries accrete', 'dictionary entries validate' ) 

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

378 

379 def __init__( 

380 self, 

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

382 /, 

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

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

385 ) -> None: 

386 self._validator_ = validator 

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

388 

389 def __repr__( self ) -> str: 

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

391 fqname = __.ccutils.qualify_class_name( type( self ) ), 

392 validator = self._validator_, 

393 contents = str( self._data_ ) ) 

394 

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

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

397 from .exceptions import EntryInvalidity 

398 raise EntryInvalidity( key, value ) 

399 return key, value 

400 

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

402 ''' Provides fresh copy of dictionary. ''' 

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

404 return dictionary.update( self ) 

405 

406 def with_data( 

407 self, 

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

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

410 ) -> __.typx.Self: 

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

412 

413 

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

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

416 

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

418 

419 _dynadoc_fragments_ = ( 

420 'dictionary entries accrete', 

421 'dictionary entries produce', 

422 'dictionary entries validate' ) 

423 _producer_: __.DictionaryProducer[ __.V ] 

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

425 

426 def __init__( 

427 self, 

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

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

430 /, 

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

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

433 ) -> None: 

434 self._producer_ = producer 

435 self._validator_ = validator 

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

437 

438 def __repr__( self ) -> str: 

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

440 fqname = __.ccutils.qualify_class_name( type( self ) ), 

441 producer = self._producer_, 

442 validator = self._validator_, 

443 contents = str( self._data_ ) ) 

444 

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

446 if key not in self: 

447 value = self._producer_( ) 

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

449 from .exceptions import EntryInvalidity 

450 raise EntryInvalidity( key, value ) 

451 self[ key ] = value 

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

453 return value 

454 

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

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

457 from .exceptions import EntryInvalidity 

458 raise EntryInvalidity( key, value ) 

459 return key, value 

460 

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

462 ''' Provides fresh copy of dictionary. ''' 

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

464 return dictionary.update( self ) 

465 

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

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

468 if key not in self: 

469 self[ key ] = default 

470 return default 

471 return self[ key ] 

472 

473 def with_data( 

474 self, 

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

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

477 ) -> __.typx.Self: 

478 return type( self )( 

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