Coverage for python/lsst/analysis/tools/interfaces/datastore/_sasquatchDatastore.py: 68%

1# This file is part of analysis_tools. 

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__ = ("SasquatchDatastore",) 

25 

26import logging 

27import os 

28from collections.abc import Collection, Iterable, Mapping, Sequence 

29from typing import TYPE_CHECKING, Any, ClassVar 

30 

31from lsst.daf.butler import DatasetRef, DatasetTypeNotSupportedError, StorageClass 

32from lsst.daf.butler.datastore import DatasetRefURIs, DatastoreConfig, DatastoreOpaqueTable 

33from lsst.daf.butler.datastore.generic_base import GenericBaseDatastore 

34from lsst.daf.butler.datastore.record_data import DatastoreRecordData 

35from lsst.daf.butler.registry.interfaces import DatastoreRegistryBridge 

36from lsst.resources import ResourcePath, ResourcePathExpression 

37 

38from . import SasquatchDispatcher, SasquatchDispatchFailure, SasquatchDispatchPartialFailure 

39from ._dispatcher import DEFAULT_TIMEOUT 

40 

41if TYPE_CHECKING: 

42 from lsst.daf.butler import Config, DatasetProvenance, DatasetType, LookupKey 

43 from lsst.daf.butler.registry.interfaces import DatasetIdRef, DatastoreRegistryBridgeManager 

44 

45"""Sasquatch datastore""" 

46 

47log = logging.getLogger(__name__) 

48 

49 

50class SasquatchDatastore(GenericBaseDatastore): 

51 """Basic Datastore for writing to an in Sasquatch instance. 

52 

53 This Datastore is currently write only, meaning that it can dispatch data 

54 to a Sasquatch instance, but at the present can not be used to retrieve 

55 values. 

56 

57 

58 Parameters 

59 ---------- 

60 config : `DatastoreConfig` or `str` 

61 Configuration. 

62 bridgeManager : `DatastoreRegistryBridgeManager` 

63 Object that manages the interface between `Registry` and datastores. 

64 butlerRoot : `str`, optional 

65 Unused parameter. 

66 """ 

67 

68 defaultConfigFile: ClassVar[str | None] = "sasquatchDatastore.yaml" 

69 """Path to configuration defaults. Accessed within the ``configs`` resource 

70 or relative to a search path. Can be None if no defaults specified. 

71 """ 

72 

73 restProxyUrl: str 

74 """Url which points to the http rest proxy. This is where datasets will be 

75 dispatched to. 

76 """ 

77 

78 accessToken: str 

79 """Access token which is used to authenticate to the restProxy. 

80 """ 

81 

82 namespace: str 

83 """The namespace in Sasquatch where the uploaded metrics will be 

84 dispatched. 

85 

86 The namespace can be read from the environment using the 

87 ``$DAF_BUTLER_SASQUATCH_NAMESPACE`` environment variable. If that is not 

88 set the datastore config ``"namespace"`` field will be checked. A default 

89 value of "lsst.dm" is used if no other value can be obtained. 

90 """ 

91 

92 extra_fields: dict[str, str | int | float] | None 

93 """Extra key/value pairs that should be passed along with the metric 

94 when storing in Sasquatch. 

95 

96 Extra fields can be obtained both from the ``$SASQUATCH_EXTRAS`` 

97 environment variable and the `"extra_fields"` entry in the datastore 

98 config. The two sources of information are merged with the environment 

99 variable taking priority. The environment variable must have the form of 

100 "k1=v1;k2=v2". 

101 """ 

102 

103 timeout: float | None 

104 """Timeout in seconds for each HTTP request made to the rest proxy, or 

105 `None` for no timeout. 

106 

107 Read from the ``"timeout"`` entry of the datastore config, defaulting to 

108 `DEFAULT_TIMEOUT`. 

109 """ 

110 

111 def __init__( 

112 self, 

113 config: DatastoreConfig, 

114 bridgeManager: DatastoreRegistryBridgeManager, 

115 butlerRoot: str | None = None, 

116 ): 

117 super().__init__(config, bridgeManager) 

118 

119 # Name ourselves either using an explicit name or a name 

120 # derived from the (unexpanded) root. 

121 self.name = self.config.get("name", "{}@{}".format(type(self).__name__, self.config["restProxyUrl"])) 

122 log.debug("Creating datastore %s", self.name) 

123 

124 self._bridge = bridgeManager.register(self.name, ephemeral=False) 

125 

126 self.restProxyUrl = self.config["restProxyUrl"] 

127 

128 self.accessToken = self.config.get("accessToken", "na") 

129 

130 self.namespace = os.environ.get( 

131 "DAF_BUTLER_SASQUATCH_NAMESPACE", # Prioritize the environment 

132 self.config.get( 

133 "namespace", # Fallback to datastore config 

134 "lsst.dm", 

135 ), 

136 ) 

137 

138 extra_fields: dict[str, str | int | float] | None = self.config.get("extra_fields", {}) 

139 if extras_str := os.environ.get("SASQUATCH_EXTRAS"): 139 ↛ 140line 139 didn't jump to line 140 because the condition on line 139 was never true

140 for item in extras_str.split(";"): 

141 k, v = item.split("=") 

142 extra_fields[k] = v 

143 self.extra_fields = extra_fields if extra_fields else None 

144 

145 self.timeout = self.config.get("timeout", DEFAULT_TIMEOUT) 

146 

147 self._dispatcher = SasquatchDispatcher( 

148 self.restProxyUrl, self.accessToken, self.namespace, timeout=self.timeout 

149 ) 

150 

151 @classmethod 

152 def _create_from_config( 

153 cls, 

154 config: DatastoreConfig, 

155 bridgeManager: DatastoreRegistryBridgeManager, 

156 butlerRoot: ResourcePathExpression | None, 

157 ) -> SasquatchDatastore: 

158 return SasquatchDatastore(config, bridgeManager) 

159 

160 def clone(self, bridgeManager: DatastoreRegistryBridgeManager) -> SasquatchDatastore: 

161 return SasquatchDatastore(self.config, bridgeManager) 

162 

163 @property 

164 def bridge(self) -> DatastoreRegistryBridge: 

165 return self._bridge 

166 

167 def put( 

168 self, inMemoryDataset: Any, ref: DatasetRef, *, provenance: DatasetProvenance | None = None 

169 ) -> None: 

170 if self.constraints.isAcceptable(ref): 170 ↛ 178line 170 didn't jump to line 178 because the condition on line 170 was always true

171 try: 

172 self._dispatcher.dispatchRef(inMemoryDataset, ref, extraFields=self.extra_fields) 

173 except SasquatchDispatchFailure: 

174 log.warning("Failed to dispatch metric bundle to Sasquatch.") 

175 except SasquatchDispatchPartialFailure: 

176 log.warning("Only some of the metrics were successfully dispatched to Sasquatch.") 

177 else: 

178 log.debug("Could not put dataset type %s with Sasquatch datastore", ref.datasetType) 

179 raise DatasetTypeNotSupportedError( 

180 f"Could not put dataset type {ref.datasetType} with Sasquatch datastore" 

181 ) 

182 

183 def export_predicted_records(self, refs: Iterable[DatasetRef]) -> dict[str, DatastoreRecordData]: 

184 return {} 

185 

186 def addStoredItemInfo(self, refs: Iterable[DatasetRef], infos: Iterable[Any]) -> None: 

187 raise NotImplementedError() 

188 

189 def getStoredItemsInfo(self, ref: DatasetRef) -> Sequence[Any]: 

190 raise NotImplementedError() 

191 

192 def removeStoredItemInfo(self, ref: DatasetRef) -> None: 

193 raise NotImplementedError() 

194 

195 def trash(self, ref: DatasetRef | Iterable[DatasetRef], ignore_errors: bool = True) -> None: 

196 log.debug("Sasquatch datastore does not support trashing skipping %s", ref) 

197 raise FileNotFoundError() 

198 

199 def emptyTrash( 

200 self, ignore_errors: bool = True, refs: Collection[DatasetRef] | None = None, dry_run: bool = False 

201 ) -> set[ResourcePath]: 

202 log.debug("Sasquatch datastore does not support trash, nothing to empty") 

203 return set() 

204 

205 def forget(self, ref: Iterable[DatasetRef]) -> None: 

206 pass 

207 

208 def exists(self, datasetRef: DatasetRef) -> bool: 

209 # sasquatch is not currently searchable 

210 return False 

211 

212 def knows(self, ref: DatasetRef) -> bool: 

213 return False 

214 

215 def get( 

216 self, 

217 datasetRef: DatasetRef, 

218 parameters: Mapping[str, Any] | None = None, 

219 storageClass: StorageClass | str | None = None, 

220 ) -> Any: 

221 raise FileNotFoundError() 

222 

223 def validateConfiguration( 

224 self, entities: Iterable[DatasetRef | DatasetType | StorageClass], logFailures: bool = False 

225 ) -> None: 

226 """Validate some of the configuration for this datastore. 

227 

228 Parameters 

229 ---------- 

230 entities : iterable of `DatasetRef`, `DatasetType`, or `StorageClass` 

231 Entities to test against this configuration. Can be differing 

232 types. 

233 logFailures : `bool`, optional 

234 If `True`, output a log message for every validation error 

235 detected. 

236 

237 Raises 

238 ------ 

239 DatastoreValidationError 

240 Raised if there is a validation problem with a configuration. 

241 All the problems are reported in a single exception. 

242 

243 Notes 

244 ----- 

245 This method is a no-op. 

246 """ 

247 return 

248 

249 def validateKey(self, lookupKey: LookupKey, entity: DatasetRef | DatasetType | StorageClass) -> None: 

250 # Docstring is inherited from base class. 

251 return 

252 

253 def getLookupKeys(self) -> set[LookupKey]: 

254 # Docstring is inherited from base class. 

255 return self.constraints.getLookupKeys() 

256 

257 def needs_expanded_data_ids( 

258 self, 

259 transfer: str | None, 

260 entity: DatasetRef | DatasetType | StorageClass | None = None, 

261 ) -> bool: 

262 # Docstring inherited. 

263 return False 

264 

265 def import_records(self, data: Mapping[str, DatastoreRecordData]) -> None: 

266 # Docstring inherited from the base class. 

267 return 

268 

269 def export_records(self, refs: Iterable[DatasetIdRef]) -> Mapping[str, DatastoreRecordData]: 

270 # Docstring inherited from the base class. 

271 

272 # Sasquatch Datastore records cannot be exported or imported. 

273 return {} 

274 

275 def getURI(self, datasetRef: DatasetRef, predict: bool = False) -> ResourcePath: 

276 raise NotImplementedError() 

277 

278 def getURIs(self, datasetRef: DatasetRef, predict: bool = False) -> DatasetRefURIs: 

279 raise NotImplementedError() 

280 

281 def ingest_zip(self, zip_path: ResourcePath, transfer: str | None) -> None: 

282 raise NotImplementedError() 

283 

284 def retrieveArtifacts( 

285 self, 

286 refs: Iterable[DatasetRef], 

287 destination: ResourcePath, 

288 transfer: str = "auto", 

289 preserve_path: bool = True, 

290 overwrite: bool = False, 

291 ) -> list[ResourcePath]: 

292 raise NotImplementedError() 

293 

294 @classmethod 

295 def setConfigRoot(cls, root: str, config: Config, full: Config, overwrite: bool = True) -> None: 

296 pass 

297 

298 def get_opaque_table_definitions(self) -> Mapping[str, DatastoreOpaqueTable]: 

299 # Docstring inherited from the base class. 

300 return {}