Coverage for python / lsst / dax / apdb / apdbIndex.py: 28%

1# This file is part of dax_apdb. 

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 program is free software: you can redistribute it and/or modify 

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

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

12# (at your option) any later version. 

13# 

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

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

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

17# GNU General Public License for more details. 

18# 

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

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

21 

22from __future__ import annotations 

23 

24__all__ = ["ApdbIndex"] 

25 

26import io 

27import logging 

28import os 

29from collections.abc import Mapping 

30from typing import ClassVar 

31 

32import yaml 

33from pydantic import TypeAdapter, ValidationError 

34 

35from lsst.resources import ResourcePath 

36 

37_LOG = logging.getLogger(__name__) 

38 

39 

40class ApdbIndex: 

41 """Index of well-known Apdb instances. 

42 

43 Parameters 

44 ---------- 

45 index_path : `str`, optional 

46 Path to the index configuration file, if not provided then the value 

47 of ``DAX_APDB_INDEX_URI`` environment variable is used to locate 

48 configuration file. 

49 

50 The index is configured from a simple YAML file whose location is 

51 determined from ``DAX_APDB_INDEX_URI`` environment variable. The content 

52 of the index file is a mapping of the labels to URIs in YAML format, e.g.: 

53 

54 .. code-block:: yaml 

55 

56 dev: "/path/to/config-file.yaml" 

57 "prod/pex_config": "s3://bucket/apdb-prod.py" 

58 "prod/yaml": "s3://bucket/apdb-prod.yaml" 

59 

60 The labels in the index file consists of the label name and an optional 

61 format name separated from label by slash. `get_apdb_uri` method can 

62 use its ``format`` argument to return either a format-specific 

63 configuration or a label-only configuration if format-specific is not 

64 defined. 

65 """ 

66 

67 index_env_var: ClassVar[str] = "DAX_APDB_INDEX_URI" 

68 """The name of the environment variable containing the URI of the index 

69 configuration file. 

70 """ 

71 

72 _cache: Mapping[str, str] | None = None 

73 """Contents of the index file cached in memory.""" 

74 

75 def __init__(self, index_path: str | None = None): 

76 self._index_path = index_path 

77 

78 def _read_index(self, index_path: str | None = None) -> Mapping[str, str]: 

79 """Return contents of the index file. 

80 

81 Parameters 

82 ---------- 

83 index_path : `str`, optional 

84 Location of the index file, if not provided then default location 

85 is used. 

86 

87 Returns 

88 ------- 

89 entries : `~collections.abc.Mapping` [`str`, `str`] 

90 All known entries. Can be empty if no index can be found. 

91 

92 Raises 

93 ------ 

94 RuntimeError 

95 Raised if ``index_path`` is not provided and environment variable 

96 is not set. 

97 TypeError 

98 Raised if content of the configuration file is incorrect. 

99 """ 

100 if self._cache is not None: 

101 return self._cache 

102 if index_path is None: 

103 index_path = os.getenv(self.index_env_var) 

104 if not index_path: 

105 raise RuntimeError( 

106 f"No repository index defined, environment variable {self.index_env_var} is not set." 

107 ) 

108 index_uri = ResourcePath(index_path) 

109 _LOG.debug("Opening YAML index file: %s", index_uri.geturl()) 

110 try: 

111 content = index_uri.read() 

112 except IsADirectoryError as exc: 

113 raise FileNotFoundError(f"Index file {index_uri.geturl()} is a directory") from exc 

114 stream = io.BytesIO(content) 

115 if index_data := yaml.load(stream, Loader=yaml.SafeLoader): 

116 try: 

117 self._cache = TypeAdapter(dict[str, str]).validate_python(index_data) 

118 except ValidationError as e: 

119 raise TypeError(f"Repository index {index_uri.geturl()} not in expected format") from e 

120 else: 

121 self._cache = {} 

122 return self._cache 

123 

124 def get_apdb_uri(self, label: str, format: str | None = None) -> ResourcePath: 

125 """Return URI for APDB configuration file given its label. 

126 

127 Parameters 

128 ---------- 

129 label : `str` 

130 Label of APDB instance. 

131 format : `str` 

132 Format of the APDB configuration file, arbitrary string. This can 

133 be used to support an expected migration from pex_config to YAML 

134 configuration for APDB, code that uses pex_config could provide 

135 "pex_config" for ``format``. The actual key in the index is 

136 either a slash-separated label and format, or, if that is missing, 

137 just a label. 

138 

139 Returns 

140 ------- 

141 uri : `~lsst.resources.ResourcePath` 

142 URI for the configuration file for APDB instance. 

143 

144 Raises 

145 ------ 

146 FileNotFoundError 

147 Raised if an index is defined in the environment but it 

148 can not be found. 

149 ValueError 

150 Raised if the label is not found in the index. 

151 RuntimeError 

152 Raised if ``index_path`` is not provided and environment variable 

153 is not set. 

154 TypeError 

155 Raised if the format of the index file is incorrect. 

156 """ 

157 index = self._read_index(self._index_path) 

158 labels: list[str] = [label] 

159 if format: 

160 labels.insert(0, f"{label}/{format}") 

161 for label in labels: 

162 if (uri_str := index.get(label)) is not None: 

163 return ResourcePath(uri_str) 

164 if len(labels) == 1: 

165 message = f"Label {labels[0]} is not defined in index file" 

166 else: 

167 labels_str = ", ".join(labels) 

168 message = f"None of labels {labels_str} is defined in index file" 

169 all_labels = set(index) 

170 raise ValueError(f"{message}, labels known to index: {all_labels}") 

171 

172 def get_entries(self) -> Mapping[str, str]: 

173 """Retrieve all entries defined in index. 

174 

175 Returns 

176 ------- 

177 entries : `~collections.abc.Mapping` [`str`, `str`] 

178 All known index entries. 

179 

180 Raises 

181 ------ 

182 RuntimeError 

183 Raised if ``index_path`` is not provided and environment variable 

184 is not set. 

185 TypeError 

186 Raised if content of the configuration file is incorrect. 

187 """ 

188 return self._read_index(self._index_path)