Coverage for python / lsst / daf / butler / _named.py: 51%

200 statements  

« prev     ^ index     » next       coverage.py v7.13.5, created at 2026-04-18 08:43 +0000

1# This file is part of daf_butler. 

2# 

3# Developed for the LSST Data Management System. 

4# This product includes software developed by the LSST Project 

5# (http://www.lsst.org). 

6# See the COPYRIGHT file at the top-level directory of this distribution 

7# for details of code ownership. 

8# 

9# This software is dual licensed under the GNU General Public License and also 

10# under a 3-clause BSD license. Recipients may choose which of these licenses 

11# to use; please see the files gpl-3.0.txt and/or bsd_license.txt, 

12# respectively. If you choose the GPL option then the following text applies 

13# (but note that there is still no warranty even if you opt for BSD instead): 

14# 

15# This program is free software: you can redistribute it and/or modify 

16# it under the terms of the GNU General Public License as published by 

17# the Free Software Foundation, either version 3 of the License, or 

18# (at your option) any later version. 

19# 

20# This program is distributed in the hope that it will be useful, 

21# but WITHOUT ANY WARRANTY; without even the implied warranty of 

22# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the 

23# GNU General Public License for more details. 

24# 

25# You should have received a copy of the GNU General Public License 

26# along with this program. If not, see <http://www.gnu.org/licenses/>. 

27from __future__ import annotations 

28 

29__all__ = ( 

30 "NameLookupMapping", 

31 "NameMappingSetView", 

32 "NamedKeyDict", 

33 "NamedKeyMapping", 

34 "NamedValueAbstractSet", 

35 "NamedValueMutableSet", 

36 "NamedValueSet", 

37) 

38 

39import contextlib 

40from abc import abstractmethod 

41from collections.abc import ( 

42 ItemsView, 

43 Iterable, 

44 Iterator, 

45 KeysView, 

46 Mapping, 

47 MutableMapping, 

48 MutableSet, 

49 Set, 

50 ValuesView, 

51) 

52from types import MappingProxyType 

53from typing import Any, Protocol, TypeVar, overload 

54 

55 

56class Named(Protocol): 

57 """Protocol for objects with string name. 

58 

59 A non-inheritance interface for objects that have a string name that 

60 maps directly to their equality comparisons. 

61 """ 

62 

63 @property 

64 def name(self) -> str: 

65 pass 

66 

67 

68K = TypeVar("K", bound=Named) 

69K_co = TypeVar("K_co", bound=Named, covariant=True) 

70V = TypeVar("V") 

71V_co = TypeVar("V_co", covariant=True) 

72 

73 

74class NamedKeyMapping(Mapping[K, V_co]): 

75 """Custom mapping class. 

76 

77 An abstract base class for custom mappings whose keys are objects with 

78 a `str` ``name`` attribute, for which lookups on the name as well as the 

79 object are permitted. 

80 

81 Notes 

82 ----- 

83 In addition to the new `names` property and `byName` method, this class 

84 simply redefines the type signature for `__getitem__` and `get` that would 

85 otherwise be inherited from `~collections.abc.Mapping`. That is only 

86 relevant for static type checking; the actual Python runtime doesn't 

87 care about types at all. 

88 """ 

89 

90 __slots__ = () 

91 

92 @property 

93 @abstractmethod 

94 def names(self) -> Set[str]: 

95 """Return the set of names associated with the keys, in the same order. 

96 

97 (`~collections.abc.Set` [ `str` ]). 

98 """ 

99 raise NotImplementedError() 

100 

101 def byName(self) -> dict[str, V_co]: 

102 """Return a `~collections.abc.Mapping` with names as keys and the 

103 ``self`` values. 

104 

105 Returns 

106 ------- 

107 dictionary : `dict` 

108 A dictionary with the same values (and iteration order) as 

109 ``self``, with `str` names as keys. This is always a new object, 

110 not a view. 

111 """ 

112 return dict(zip(self.names, self.values(), strict=True)) 

113 

114 @abstractmethod 

115 def keys(self) -> NamedValueAbstractSet[K]: # type: ignore 

116 # TODO: docs 

117 raise NotImplementedError() 

118 

119 @abstractmethod 

120 def __getitem__(self, key: str | K) -> V_co: 

121 raise NotImplementedError() 

122 

123 @overload 

124 def get(self, key: object) -> V_co | None: ... 124 ↛ exitline 124 didn't return from function 'get' because

125 

126 @overload 

127 def get(self, key: object, default: V) -> V_co | V: ... 127 ↛ exitline 127 didn't return from function 'get' because

128 

129 def get(self, key: Any, default: Any = None) -> Any: 

130 return super().get(key, default) 

131 

132 

133NameLookupMapping = NamedKeyMapping[K, V_co] | Mapping[str, V_co] 

134"""A type annotation alias for signatures that want to use ``mapping[s]`` 

135(or ``mapping.get(s)``) where ``s`` is a `str`, and don't care whether 

136``mapping.keys()`` returns named objects or direct `str` instances. 

137""" 

138 

139 

140class NamedKeyMutableMapping(NamedKeyMapping[K, V], MutableMapping[K, V]): 

141 """An abstract base class that adds mutation to `NamedKeyMapping`.""" 

142 

143 __slots__ = () 

144 

145 @abstractmethod 

146 def __setitem__(self, key: str | K, value: V) -> None: 

147 raise NotImplementedError() 

148 

149 @abstractmethod 

150 def __delitem__(self, key: str | K) -> None: 

151 raise NotImplementedError() 

152 

153 def pop(self, key: str | K, default: Any = None) -> Any: 

154 # See comment in `NamedKeyMapping.get`; same logic applies here. 

155 return super().pop(key, default) # type: ignore 

156 

157 

158class NamedKeyDict(NamedKeyMutableMapping[K, V]): 

159 """Dictionary wrapper for named keys. 

160 

161 Requires keys to have a ``.name`` attribute, 

162 and permits lookups using either key objects or their names. 

163 

164 Names can be used in place of keys when updating existing items, but not 

165 when adding new items. 

166 

167 It is assumed (but asserted) that all name equality is equivalent to key 

168 equality, either because the key objects define equality this way, or 

169 because different objects with the same name are never included in the same 

170 dictionary. 

171 

172 Parameters 

173 ---------- 

174 *args : `typing.Any` 

175 All positional constructor arguments are forwarded directly to `dict`. 

176 Keyword arguments are not accepted, because plain strings are not valid 

177 keys for `NamedKeyDict`. 

178 

179 Raises 

180 ------ 

181 AttributeError 

182 Raised when an attempt is made to add an object with no ``.name`` 

183 attribute to the dictionary. 

184 AssertionError 

185 Raised when multiple keys have the same name. 

186 """ 

187 

188 __slots__ = ( 

189 "_dict", 

190 "_names", 

191 ) 

192 

193 def __init__(self, *args: Any): 

194 self._dict: dict[K, V] = dict(*args) 

195 self._names = {key.name: key for key in self._dict} 

196 assert len(self._names) == len(self._dict), "Duplicate names in keys." 

197 

198 @property 

199 def names(self) -> KeysView[str]: 

200 """Return set of names associated with the keys, in the same order. 

201 

202 (`~collections.abc.KeysView`). 

203 """ 

204 return self._names.keys() 

205 

206 def byName(self) -> dict[str, V]: 

207 """Return a `dict` with names as keys and the ``self`` values.""" 

208 return dict(zip(self._names.keys(), self._dict.values(), strict=True)) 

209 

210 def __len__(self) -> int: 

211 return len(self._dict) 

212 

213 def __iter__(self) -> Iterator[K]: 

214 return iter(self._dict) 

215 

216 def __str__(self) -> str: 

217 return "{{{}}}".format(", ".join(f"{str(k)}: {str(v)}" for k, v in self.items())) 

218 

219 def __repr__(self) -> str: 

220 return "NamedKeyDict({{{}}})".format(", ".join(f"{repr(k)}: {repr(v)}" for k, v in self.items())) 

221 

222 def __getitem__(self, key: str | K) -> V: 

223 if isinstance(key, str): 

224 return self._dict[self._names[key]] 

225 else: 

226 return self._dict[key] 

227 

228 def __setitem__(self, key: str | K, value: V) -> None: 

229 if isinstance(key, str): 

230 self._dict[self._names[key]] = value 

231 else: 

232 assert self._names.get(key.name, key) == key, "Name is already associated with a different key." 

233 self._dict[key] = value 

234 self._names[key.name] = key 

235 

236 def __delitem__(self, key: str | K) -> None: 

237 if isinstance(key, str): 

238 del self._dict[self._names[key]] 

239 del self._names[key] 

240 else: 

241 del self._dict[key] 

242 del self._names[key.name] 

243 

244 def keys(self) -> NamedValueAbstractSet[K]: # type: ignore 

245 return NameMappingSetView(self._names) 

246 

247 def values(self) -> ValuesView[V]: 

248 return self._dict.values() 

249 

250 def items(self) -> ItemsView[K, V]: 

251 return self._dict.items() 

252 

253 def copy(self) -> NamedKeyDict[K, V]: 

254 """Return a new `NamedKeyDict` with the same elements.""" 

255 result = NamedKeyDict.__new__(NamedKeyDict) 

256 result._dict = dict(self._dict) 

257 result._names = dict(self._names) 

258 return result 

259 

260 def freeze(self) -> NamedKeyMapping[K, V]: 

261 """Disable all mutators. 

262 

263 Effectively transforms ``self`` into an immutable mapping. 

264 

265 Returns 

266 ------- 

267 self : `NamedKeyMapping` 

268 While ``self`` is modified in-place, it is also returned with a 

269 type annotation that reflects its new, frozen state; assigning it 

270 to a new variable (and considering any previous references 

271 invalidated) should allow for more accurate static type checking. 

272 """ 

273 if not isinstance(self._dict, MappingProxyType): # type: ignore[unreachable] 

274 self._dict = MappingProxyType(self._dict) # type: ignore 

275 return self 

276 

277 

278class NamedValueAbstractSet(Set[K_co]): 

279 """Custom sets with named elements. 

280 

281 An abstract base class for custom sets whose elements are objects with 

282 a `str` ``name`` attribute, allowing some dict-like operations and 

283 views to be supported. 

284 """ 

285 

286 __slots__ = () 

287 

288 @property 

289 @abstractmethod 

290 def names(self) -> Set[str]: 

291 """Return set of names associated with the keys, in the same order. 

292 

293 (`~collections.abc.Set` [ `str` ]). 

294 """ 

295 raise NotImplementedError() 

296 

297 @abstractmethod 

298 def asMapping(self) -> Mapping[str, K_co]: 

299 """Return a mapping view with names as keys. 

300 

301 Returns 

302 ------- 

303 dict : `~collections.abc.Mapping` 

304 A dictionary-like view with ``values() == self``. 

305 """ 

306 raise NotImplementedError() 

307 

308 @abstractmethod 

309 def __getitem__(self, key: str | K_co) -> K_co: 

310 raise NotImplementedError() 

311 

312 @overload 

313 def get(self, key: object) -> K_co | None: ... 313 ↛ exitline 313 didn't return from function 'get' because

314 

315 @overload 

316 def get(self, key: object, default: V) -> K_co | V: ... 316 ↛ exitline 316 didn't return from function 'get' because

317 

318 def get(self, key: Any, default: Any = None) -> Any: 

319 """Return the element with the given name. 

320 

321 Parameters 

322 ---------- 

323 key : `typing.Any` 

324 The name of the element to be requested. 

325 default : `typing.Any`, optional 

326 The value returned if no such element is present. 

327 

328 Returns 

329 ------- 

330 result : `typing.Any` 

331 The value of the element. 

332 """ 

333 try: 

334 return self[key] 

335 except KeyError: 

336 return default 

337 

338 # MyPy wants _from_iterable to be fully generic and work on all types, 

339 # but a NamedValueSet only wants to be iterable with other sets that have 

340 # the same item type. 

341 @classmethod 

342 def _from_iterable(cls, iterable: Iterable[K_co]) -> NamedValueSet[K_co]: # type: ignore[override] 

343 """Construct class from an iterable. 

344 

345 Hook to ensure that inherited `collections.abc.Set` operators return 

346 `NamedValueSet` instances, not something else (see `collections.abc` 

347 documentation for more information). 

348 

349 Note that this behavior can only be guaranteed when both operands are 

350 `NamedValueAbstractSet` instances. 

351 """ 

352 return NamedValueSet(iterable) 

353 

354 

355class NameMappingSetView(NamedValueAbstractSet[K_co]): 

356 """A lightweight implementation of `NamedValueAbstractSet`. 

357 

358 Backed by a mapping from name to named object. 

359 

360 Parameters 

361 ---------- 

362 mapping : `~collections.abc.Mapping` [ `str`, `object` ] 

363 Mapping this object will provide a view of. 

364 """ 

365 

366 def __init__(self, mapping: Mapping[str, K_co]): 

367 self._mapping = mapping 

368 

369 __slots__ = ("_mapping",) 

370 

371 @property 

372 def names(self) -> Set[str]: 

373 # Docstring inherited from NamedValueAbstractSet. 

374 return self._mapping.keys() 

375 

376 def asMapping(self) -> Mapping[str, K_co]: 

377 # Docstring inherited from NamedValueAbstractSet. 

378 return self._mapping 

379 

380 def __getitem__(self, key: str | K_co) -> K_co: 

381 if isinstance(key, str): 

382 return self._mapping[key] 

383 else: 

384 return self._mapping[key.name] 

385 

386 def __contains__(self, key: Any) -> bool: 

387 return getattr(key, "name", key) in self._mapping 

388 

389 def __len__(self) -> int: 

390 return len(self._mapping) 

391 

392 def __iter__(self) -> Iterator[K_co]: 

393 return iter(self._mapping.values()) 

394 

395 def __eq__(self, other: Any) -> bool: 

396 if isinstance(other, NamedValueAbstractSet): 

397 return self.names == other.names 

398 else: 

399 return set(self._mapping.values()) == other 

400 

401 def __le__(self, other: Set[K]) -> bool: 

402 if isinstance(other, NamedValueAbstractSet): 

403 return self.names <= other.names 

404 else: 

405 return set(self._mapping.values()) <= other 

406 

407 def __ge__(self, other: Set[K]) -> bool: 

408 if isinstance(other, NamedValueAbstractSet): 

409 return self.names >= other.names 

410 else: 

411 return set(self._mapping.values()) >= other 

412 

413 def __str__(self) -> str: 

414 return "{{{}}}".format(", ".join(str(element) for element in self)) 

415 

416 def __repr__(self) -> str: 

417 return f"NameMappingSetView({self._mapping})" 

418 

419 

420class NamedValueMutableSet(NamedValueAbstractSet[K], MutableSet[K]): 

421 """Mutable variant of `NamedValueAbstractSet`. 

422 

423 Methods that can add new elements to the set are unchanged from their 

424 `~collections.abc.MutableSet` definitions, while those that only remove 

425 them can generally accept names or element instances. `pop` can be used 

426 in either its `~collections.abc.MutableSet` form (no arguments; an 

427 arbitrary element is returned) or its `~collections.abc.MutableMapping` 

428 form (one or two arguments for the name and optional default value, 

429 respectively). A `~collections.abc.MutableMapping`-like `__delitem__` 

430 interface is also included, which takes only names (like 

431 `NamedValueAbstractSet.__getitem__`). 

432 """ 

433 

434 __slots__ = () 

435 

436 @abstractmethod 

437 def __delitem__(self, name: str) -> None: 

438 raise NotImplementedError() 

439 

440 @abstractmethod 

441 def remove(self, element: str | K) -> Any: 

442 """Remove an element from the set. 

443 

444 Parameters 

445 ---------- 

446 element : `object` or `str` 

447 Element to remove or the string name thereof. Assumed to be an 

448 element if it has a ``.name`` attribute. 

449 

450 Raises 

451 ------ 

452 KeyError 

453 Raised if an element with the given name does not exist. 

454 """ 

455 raise NotImplementedError() 

456 

457 @abstractmethod 

458 def discard(self, element: str | K) -> Any: 

459 """Remove an element from the set if it exists. 

460 

461 Does nothing if no matching element is present. 

462 

463 Parameters 

464 ---------- 

465 element : `object` or `str` 

466 Element to remove or the string name thereof. Assumed to be an 

467 element if it has a ``.name`` attribute. 

468 """ 

469 raise NotImplementedError() 

470 

471 @abstractmethod 

472 def pop(self, *args: str) -> K: 

473 """Remove and return an element from the set. 

474 

475 Parameters 

476 ---------- 

477 *args : `str`, optional 

478 Name of the element to remove and return. Must be passed 

479 positionally. If not provided, an arbitrary element is 

480 removed and returned. 

481 

482 Raises 

483 ------ 

484 KeyError 

485 Raised if ``name`` is provided but ``default`` is not, and no 

486 matching element exists. 

487 """ 

488 raise NotImplementedError() 

489 

490 

491class NamedValueSet(NameMappingSetView[K], NamedValueMutableSet[K]): 

492 """Custom mutable set class. 

493 

494 A custom mutable set class that requires elements to have a ``.name`` 

495 attribute, which can then be used as keys in `dict`-like lookup. 

496 

497 Names and elements can both be used with the ``in`` and ``del`` 

498 operators, `remove`, and `discard`. Names (but not elements) 

499 can be used with ``[]``-based element retrieval (not assignment) 

500 and the `get` method. 

501 

502 Parameters 

503 ---------- 

504 elements : `collections.abc.Iterable` 

505 Iterable over elements to include in the set. 

506 

507 Raises 

508 ------ 

509 AttributeError 

510 Raised if one or more elements do not have a ``.name`` attribute. 

511 

512 Notes 

513 ----- 

514 Iteration order is guaranteed to be the same as insertion order (with 

515 the same general behavior as `dict` ordering). 

516 Like `dict`, sets with the same elements will compare as equal even if 

517 their iterator order is not the same. 

518 """ 

519 

520 def __init__(self, elements: Iterable[K] = ()): 

521 super().__init__({element.name: element for element in elements}) 

522 

523 def __repr__(self) -> str: 

524 return "NamedValueSet({{{}}})".format(", ".join(repr(element) for element in self)) 

525 

526 def issubset(self, other: Set[K]) -> bool: 

527 return self <= other 

528 

529 def issuperset(self, other: Set[K]) -> bool: 

530 return self >= other 

531 

532 def __delitem__(self, name: str) -> None: 

533 del self._mapping[name] 

534 

535 def add(self, element: K) -> None: 

536 """Add an element to the set. 

537 

538 Parameters 

539 ---------- 

540 element : `typing.Any` 

541 The element to add. 

542 

543 Raises 

544 ------ 

545 AttributeError 

546 Raised if the element does not have a ``.name`` attribute. 

547 """ 

548 self._mapping[element.name] = element 

549 

550 def clear(self) -> None: 

551 # Docstring inherited. 

552 self._mapping.clear() 

553 

554 def remove(self, element: str | K) -> Any: 

555 # Docstring inherited. 

556 k = element.name if not isinstance(element, str) else element 

557 del self._mapping[k] 

558 

559 def discard(self, element: str | K) -> Any: 

560 # Docstring inherited. 

561 with contextlib.suppress(KeyError): 

562 self.remove(element) 

563 

564 def pop(self, *args: str) -> K: 

565 # Docstring inherited. 

566 if not args: 

567 # Parent is abstract method and we cannot call MutableSet 

568 # implementation directly. Instead follow MutableSet and 

569 # choose first element from iteration. 

570 it = iter(self._mapping) 

571 try: 

572 value = next(it) 

573 except StopIteration: 

574 raise KeyError from None 

575 args = (value,) 

576 

577 return self._mapping.pop(*args) 

578 

579 def update(self, elements: Iterable[K]) -> None: 

580 """Add multiple new elements to the set. 

581 

582 Parameters 

583 ---------- 

584 elements : `~collections.abc.Iterable` 

585 Elements to add. 

586 """ 

587 for element in elements: 

588 self.add(element) 

589 

590 def copy(self) -> NamedValueSet[K]: 

591 """Return a new `NamedValueSet` with the same elements.""" 

592 result = NamedValueSet.__new__(NamedValueSet) 

593 result._mapping = dict(self._mapping) 

594 return result 

595 

596 def freeze(self) -> NamedValueAbstractSet[K]: 

597 """Disable all mutators. 

598 

599 Effectively transforming ``self`` into an immutable set. 

600 

601 Returns 

602 ------- 

603 self : `NamedValueAbstractSet` 

604 While ``self`` is modified in-place, it is also returned with a 

605 type annotation that reflects its new, frozen state; assigning it 

606 to a new variable (and considering any previous references 

607 invalidated) should allow for more accurate static type checking. 

608 """ 

609 if not isinstance(self._mapping, MappingProxyType): # type: ignore[unreachable] 

610 self._mapping = MappingProxyType(self._mapping) # type: ignore 

611 return self 

612 

613 _mapping: dict[str, K]