Coverage for python / lsst / daf / butler / dimensions / record_cache.py: 38%

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/>. 

27 

28from __future__ import annotations 

29 

30__all__ = ("DimensionRecordCache",) 

31 

32import copy 

33from collections.abc import Callable, Iterator, Mapping 

34 

35from ._record_set import DimensionRecordSet 

36from ._universe import DimensionUniverse 

37 

38 

39class DimensionRecordCache(Mapping[str, DimensionRecordSet]): 

40 """A mapping of cached dimension records. 

41 

42 This object holds all records for elements where 

43 `DimensionElement.is_cached` is `True`. 

44 

45 Parameters 

46 ---------- 

47 universe : `DimensionUniverse` 

48 Definitions of all dimensions. 

49 fetch : `~collections.abc.Callable` 

50 A callable that takes no arguments and returns a `dict` mapping `str` 

51 element name to a `DimensionRecordSet` of all records for that element. 

52 They keys of the returned `dict` must be exactly the elements in 

53 ``universe`` for which `DimensionElement.is_cached` is `True`. 

54 

55 Notes 

56 ----- 

57 The nested `DimensionRecordSet` objects should never be modified in place. 

58 """ 

59 

60 def __init__(self, universe: DimensionUniverse, fetch: Callable[[], dict[str, DimensionRecordSet]]): 

61 self._universe = universe 

62 self._keys = [element.name for element in universe.elements if element.is_cached] 

63 self._records: dict[str, DimensionRecordSet] | None = None 

64 self._fetch = fetch 

65 

66 def reset(self) -> None: 

67 """Reset the cache, causing it to be fetched again on next use.""" 

68 self._records = None 

69 

70 def load_from(self, other: DimensionRecordCache) -> None: 

71 """Load records from another cache, but do nothing if it doesn't 

72 currently have any records. 

73 

74 Parameters 

75 ---------- 

76 other : `DimensionRecordCache` 

77 Other cache to potentially copy records from. 

78 """ 

79 self._records = copy.deepcopy(other._records) 

80 

81 def preload_cache(self) -> None: 

82 """Fetch the cache from the DB if it has not already been fetched.""" 

83 if self._records is None: 

84 self._records = self._fetch() 

85 assert self._records.keys() == set(self._keys), "Logic bug in fetch callback." 

86 

87 def __contains__(self, key: object) -> bool: 

88 if not isinstance(key, str): 

89 return False 

90 if (element := self._universe.get(key)) is not None: 

91 return element.is_cached 

92 return False 

93 

94 def __getitem__(self, element: str) -> DimensionRecordSet: 

95 self.preload_cache() 

96 assert self._records is not None 

97 return self._records[element] 

98 

99 def __iter__(self) -> Iterator[str]: 

100 return iter(self._keys) 

101 

102 def __len__(self) -> int: 

103 return len(self._keys)