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

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 put_new(self, in_memory_dataset: Any, dataset_ref: DatasetRef) -> Mapping[str, DatasetRef]: 

184 # Docstring inherited from the base class. 

185 self.put(in_memory_dataset, dataset_ref) 

186 # Sasquatch is a sort of ephemeral, because we do not store its 

187 # datastore records in registry, so return empty dict. 

188 return {} 

189 

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

191 return {} 

192 

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

194 raise NotImplementedError() 

195 

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

197 raise NotImplementedError() 

198 

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

200 raise NotImplementedError() 

201 

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

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

204 raise FileNotFoundError() 

205 

206 def emptyTrash( 

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

208 ) -> set[ResourcePath]: 

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

210 return set() 

211 

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

213 pass 

214 

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

216 # sasquatch is not currently searchable 

217 return False 

218 

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

220 return False 

221 

222 def get( 

223 self, 

224 datasetRef: DatasetRef, 

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

226 storageClass: StorageClass | str | None = None, 

227 ) -> Any: 

228 raise FileNotFoundError() 

229 

230 def validateConfiguration( 

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

232 ) -> None: 

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

234 

235 Parameters 

236 ---------- 

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

238 Entities to test against this configuration. Can be differing 

239 types. 

240 logFailures : `bool`, optional 

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

242 detected. 

243 

244 Raises 

245 ------ 

246 DatastoreValidationError 

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

248 All the problems are reported in a single exception. 

249 

250 Notes 

251 ----- 

252 This method is a no-op. 

253 """ 

254 return 

255 

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

257 # Docstring is inherited from base class. 

258 return 

259 

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

261 # Docstring is inherited from base class. 

262 return self.constraints.getLookupKeys() 

263 

264 def needs_expanded_data_ids( 

265 self, 

266 transfer: str | None, 

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

268 ) -> bool: 

269 # Docstring inherited. 

270 return False 

271 

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

273 # Docstring inherited from the base class. 

274 return 

275 

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

277 # Docstring inherited from the base class. 

278 

279 # Sasquatch Datastore records cannot be exported or imported. 

280 return {} 

281 

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

283 raise NotImplementedError() 

284 

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

286 raise NotImplementedError() 

287 

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

289 raise NotImplementedError() 

290 

291 def retrieveArtifacts( 

292 self, 

293 refs: Iterable[DatasetRef], 

294 destination: ResourcePath, 

295 transfer: str = "auto", 

296 preserve_path: bool = True, 

297 overwrite: bool = False, 

298 ) -> list[ResourcePath]: 

299 raise NotImplementedError() 

300 

301 @classmethod 

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

303 pass 

304 

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

306 # Docstring inherited from the base class. 

307 return {}