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 class_mutables = _classes.abc_class_mutables, 

230): 

231 ''' Accretive dictionary. ''' 

232 

233 __slots__ = ( '_data_', ) 

234 

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

236 _dynadoc_fragments_ = ( 'dictionary entries accrete', ) 

237 

238 def __init__( 

239 self, 

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

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

242 ) -> None: 

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

244 super( ).__init__( ) 

245 

246 __hash__ = None 

247 

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

249 return iter( self._data_ ) 

250 

251 def __len__( self ) -> int: 

252 return len( self._data_ ) 

253 

254 def __repr__( self ) -> str: 

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

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

257 contents = str( self._data_ ) ) 

258 

259 def __str__( self ) -> str: 

260 return str( self._data_ ) 

261 

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

263 return key in self._data_ 

264 

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

266 return self._data_[ key ] 

267 

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

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

270 return self._data_ == other 

271 return NotImplemented 

272 

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

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

275 return self._data_ != other 

276 return NotImplemented 

277 

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

279 ''' Provides fresh copy of dictionary. ''' 

280 return type( self )( self ) 

281 

282 def get( # pyright: ignore 

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

284 ) -> __.typx.Annotated[ 

285 __.V, 

286 __.typx.Doc( 

287 'Value of entry, if it exists. ' 

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

289 ]: 

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

291 if __.is_absent( default ): 

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

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

294 

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

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

297 return self._data_.keys( ) 

298 

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

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

301 return self._data_.items( ) 

302 

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

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

305 return self._data_.values( ) 

306 

307 def with_data( 

308 self, 

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

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

311 ) -> __.typx.Self: 

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

313 

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

315 self._data_[ key ] = value 

316 

317 

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

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

320 

321 __slots__ = ( '_producer_', ) 

322 

323 _dynadoc_fragments_ = ( 

324 'dictionary entries accrete', 'dictionary entries produce' ) 

325 _producer_: __.DictionaryProducer[ __.V ] 

326 

327 def __init__( 

328 self, 

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

330 /, 

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

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

333 ): 

334 # TODO: Validate producer argument. 

335 self._producer_ = producer 

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

337 

338 def __repr__( self ) -> str: 

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

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

341 producer = self._producer_, 

342 contents = str( self._data_ ) ) 

343 

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

345 if key not in self: 

346 value = self._producer_( ) 

347 self[ key ] = value 

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

349 return value 

350 

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

352 ''' Provides fresh copy of dictionary. ''' 

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

354 return dictionary.update( self ) 

355 

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

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

358 if key not in self: 

359 self[ key ] = default 

360 return default 

361 return self[ key ] 

362 

363 def with_data( 

364 self, 

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

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

367 ) -> __.typx.Self: 

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

369 

370 

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

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

373 

374 __slots__ = ( '_validator_', ) 

375 

376 _dynadoc_fragments_ = ( 

377 'dictionary entries accrete', 'dictionary entries validate' ) 

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

379 

380 def __init__( 

381 self, 

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

383 /, 

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

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

386 ) -> None: 

387 self._validator_ = validator 

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

389 

390 def __repr__( self ) -> str: 

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

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

393 validator = self._validator_, 

394 contents = str( self._data_ ) ) 

395 

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

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

398 from .exceptions import EntryInvalidity 

399 raise EntryInvalidity( key, value ) 

400 return key, value 

401 

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

403 ''' Provides fresh copy of dictionary. ''' 

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

405 return dictionary.update( self ) 

406 

407 def with_data( 

408 self, 

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

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

411 ) -> __.typx.Self: 

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

413 

414 

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

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

417 

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

419 

420 _dynadoc_fragments_ = ( 

421 'dictionary entries accrete', 

422 'dictionary entries produce', 

423 'dictionary entries validate' ) 

424 _producer_: __.DictionaryProducer[ __.V ] 

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

426 

427 def __init__( 

428 self, 

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

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

431 /, 

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

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

434 ) -> None: 

435 self._producer_ = producer 

436 self._validator_ = validator 

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

438 

439 def __repr__( self ) -> str: 

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

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

442 producer = self._producer_, 

443 validator = self._validator_, 

444 contents = str( self._data_ ) ) 

445 

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

447 if key not in self: 

448 value = self._producer_( ) 

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

450 from .exceptions import EntryInvalidity 

451 raise EntryInvalidity( key, value ) 

452 self[ key ] = value 

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

454 return value 

455 

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

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

458 from .exceptions import EntryInvalidity 

459 raise EntryInvalidity( key, value ) 

460 return key, value 

461 

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

463 ''' Provides fresh copy of dictionary. ''' 

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

465 return dictionary.update( self ) 

466 

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

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

469 if key not in self: 

470 self[ key ] = default 

471 return default 

472 return self[ key ] 

473 

474 def with_data( 

475 self, 

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

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

478 ) -> __.typx.Self: 

479 return type( self )( 

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