Coverage for python/lsst/daf/butler/remote_butler/_registry.py: 45%

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 

30import contextlib 

31from collections.abc import Iterable, Iterator, Mapping, Sequence 

32from typing import Any 

33 

34from lsst.daf.butler import Butler 

35from lsst.utils.iteration import ensure_iterable 

36 

37from .._collection_type import CollectionType 

38from .._dataset_ref import DatasetId, DatasetIdGenEnum, DatasetRef 

39from .._dataset_type import DatasetType 

40from .._storage_class import StorageClassFactory 

41from .._timespan import Timespan 

42from ..dimensions import ( 

43 DataCoordinate, 

44 DataId, 

45 DimensionElement, 

46 DimensionGroup, 

47 DimensionRecord, 

48 DimensionUniverse, 

49) 

50from ..registry import ( 

51 CollectionArgType, 

52 CollectionSummary, 

53 CollectionTypeError, 

54 DatasetTypeError, 

55 RegistryDefaults, 

56) 

57from ..registry._registry_base import RegistryBase 

58from ._collection_args import ( 

59 convert_collection_arg_to_glob_string_list, 

60 convert_dataset_type_arg_to_glob_string_list, 

61) 

62from ._defaults import DefaultsHolder 

63from ._http_connection import RemoteButlerHttpConnection, parse_model 

64from .server_models import ( 

65 ExpandDataIdRequestModel, 

66 ExpandDataIdResponseModel, 

67 GetCollectionSummaryResponseModel, 

68 QueryDatasetTypesRequestModel, 

69 QueryDatasetTypesResponseModel, 

70) 

71 

72 

73class RemoteButlerRegistry(RegistryBase): 

74 """Implementation of the `Registry` interface for `RemoteButler. 

75 

76 Parameters 

77 ---------- 

78 butler : `Butler` 

79 Butler instance to which this registry delegates operations. 

80 defaults : `DefaultHolder` 

81 Reference to object containing default collections and data ID. 

82 connection : `RemoteButlerHttpConnection` 

83 HTTP connection to Butler server for looking up data. 

84 """ 

85 

86 def __init__(self, butler: Butler, defaults: DefaultsHolder, connection: RemoteButlerHttpConnection): 

87 super().__init__(butler) 

88 self._connection = connection 

89 self._defaults = defaults 

90 

91 def isWriteable(self) -> bool: 

92 return self._butler.isWriteable() 

93 

94 @property 

95 def dimensions(self) -> DimensionUniverse: 

96 return self._butler.dimensions 

97 

98 @property 

99 def defaults(self) -> RegistryDefaults: 

100 return self._defaults.get() 

101 

102 @defaults.setter 

103 def defaults(self, value: RegistryDefaults) -> None: 

104 value.finish(self) 

105 self._defaults.set(value) 

106 

107 def refresh(self) -> None: 

108 # In theory the server should manage all necessary invalidation of 

109 # state. 

110 pass 

111 

112 def refresh_collection_summaries(self) -> None: 

113 # Docstring inherited from a base class. 

114 raise NotImplementedError() 

115 

116 @contextlib.contextmanager 

117 def caching_context(self) -> Iterator[None]: 

118 yield 

119 

120 @contextlib.contextmanager 

121 def transaction(self, *, savepoint: bool = False) -> Iterator[None]: 

122 # RemoteButler will never support transactions 

123 raise NotImplementedError() 

124 

125 def registerCollection( 

126 self, name: str, type: CollectionType = CollectionType.TAGGED, doc: str | None = None 

127 ) -> bool: 

128 raise NotImplementedError() 

129 

130 def getCollectionType(self, name: str) -> CollectionType: 

131 return self._butler.collections.get_info(name).type 

132 

133 def registerRun(self, name: str, doc: str | None = None) -> bool: 

134 raise NotImplementedError() 

135 

136 def removeCollection(self, name: str) -> None: 

137 raise NotImplementedError() 

138 

139 def getCollectionChain(self, parent: str) -> Sequence[str]: 

140 info = self._butler.collections.get_info(parent) 

141 if info.type is not CollectionType.CHAINED: 

142 raise CollectionTypeError(f"Collection '{parent}' has type {info.type.name}, not CHAINED.") 

143 return info.children 

144 

145 def setCollectionChain(self, parent: str, children: Any, *, flatten: bool = False) -> None: 

146 raise NotImplementedError() 

147 

148 def getCollectionParentChains(self, collection: str) -> set[str]: 

149 info = self._butler.collections.get_info(collection, include_parents=True) 

150 assert info.parents is not None, "Requested list of parents from server, but it did not send them." 

151 return set(info.parents) 

152 

153 def getCollectionDocumentation(self, collection: str) -> str | None: 

154 doc = self._butler.collections.get_info(collection).doc 

155 if not doc: 

156 return None 

157 return doc 

158 

159 def setCollectionDocumentation(self, collection: str, doc: str | None) -> None: 

160 raise NotImplementedError() 

161 

162 def getCollectionSummary(self, collection: str) -> CollectionSummary: 

163 response = self._connection.get("collection_summary", {"name": collection}) 

164 parsed = parse_model(response, GetCollectionSummaryResponseModel) 

165 return CollectionSummary.from_simple(parsed.summary, self.dimensions) 

166 

167 def registerDatasetType(self, datasetType: DatasetType) -> bool: 

168 raise NotImplementedError() 

169 

170 def removeDatasetType(self, name: str | tuple[str, ...]) -> None: 

171 raise NotImplementedError() 

172 

173 def getDatasetType(self, name: str) -> DatasetType: 

174 return self._butler.get_dataset_type(name) 

175 

176 def supportsIdGenerationMode(self, mode: DatasetIdGenEnum) -> bool: 

177 raise NotImplementedError() 

178 

179 def findDataset( 

180 self, 

181 datasetType: DatasetType | str, 

182 dataId: DataId | None = None, 

183 *, 

184 collections: CollectionArgType | None = None, 

185 timespan: Timespan | None = None, 

186 datastore_records: bool = False, 

187 **kwargs: Any, 

188 ) -> DatasetRef | None: 

189 # Components are supported by Butler.find_dataset, but are required to 

190 # raise an exception in registry.findDataset (see 

191 # RegistryTests.testComponentLookups). Apparently the registry version 

192 # used to support components, but at some point a decision was made to 

193 # draw a hard architectural boundary between registry and datastore so 

194 # that registry is no longer allowed to know about components. 

195 if _is_component_dataset_type(datasetType): 

196 raise DatasetTypeError( 

197 "Component dataset types are not supported;" 

198 " use DatasetRef or DatasetType methods to obtain components from parents instead." 

199 ) 

200 

201 if collections is not None: 

202 collections = convert_collection_arg_to_glob_string_list(collections) 

203 

204 return self._butler.find_dataset( 

205 datasetType, 

206 dataId, 

207 collections=collections, 

208 timespan=timespan, 

209 datastore_records=datastore_records, 

210 **kwargs, 

211 ) 

212 

213 def insertDatasets( 

214 self, 

215 datasetType: DatasetType | str, 

216 dataIds: Iterable[DataId], 

217 run: str | None = None, 

218 expand: bool = True, 

219 idGenerationMode: DatasetIdGenEnum = DatasetIdGenEnum.UNIQUE, 

220 ) -> list[DatasetRef]: 

221 raise NotImplementedError() 

222 

223 def _importDatasets( 

224 self, 

225 datasets: Iterable[DatasetRef], 

226 expand: bool = True, 

227 ) -> list[DatasetRef]: 

228 raise NotImplementedError() 

229 

230 def getDataset(self, id: DatasetId) -> DatasetRef | None: 

231 return self._butler.get_dataset(id) 

232 

233 def removeDatasets(self, refs: Iterable[DatasetRef]) -> None: 

234 raise NotImplementedError() 

235 

236 def associate(self, collection: str, refs: Iterable[DatasetRef]) -> None: 

237 raise NotImplementedError() 

238 

239 def disassociate(self, collection: str, refs: Iterable[DatasetRef]) -> None: 

240 raise NotImplementedError() 

241 

242 def certify(self, collection: str, refs: Iterable[DatasetRef], timespan: Timespan) -> None: 

243 raise NotImplementedError() 

244 

245 def decertify( 

246 self, 

247 collection: str, 

248 datasetType: str | DatasetType, 

249 timespan: Timespan, 

250 *, 

251 dataIds: Iterable[DataId] | None = None, 

252 ) -> None: 

253 raise NotImplementedError() 

254 

255 def getDatasetLocations(self, ref: DatasetRef) -> Iterable[str]: 

256 raise NotImplementedError() 

257 

258 def expandDataId( 

259 self, 

260 dataId: DataId | None = None, 

261 *, 

262 dimensions: Iterable[str] | DimensionGroup | None = None, 

263 records: Mapping[str, DimensionRecord | None] | None = None, 

264 withDefaults: bool = True, 

265 **kwargs: Any, 

266 ) -> DataCoordinate: 

267 standardized = DataCoordinate.standardize( 

268 dataId, 

269 dimensions=dimensions, 

270 universe=self.dimensions, 

271 defaults=self.defaults.dataId if withDefaults else None, 

272 **kwargs, 

273 ) 

274 if standardized.hasRecords(): 

275 return standardized 

276 

277 request = ExpandDataIdRequestModel(data_id=standardized.to_simple().dataId) 

278 response = self._connection.post("expand_data_id", request) 

279 model = parse_model(response, ExpandDataIdResponseModel) 

280 return DataCoordinate.from_simple(model.data_coordinate, self.dimensions) 

281 

282 def insertDimensionData( 

283 self, 

284 element: DimensionElement | str, 

285 *data: Mapping[str, Any] | DimensionRecord, 

286 conform: bool = True, 

287 replace: bool = False, 

288 skip_existing: bool = False, 

289 ) -> None: 

290 raise NotImplementedError() 

291 

292 def syncDimensionData( 

293 self, 

294 element: DimensionElement | str, 

295 row: Mapping[str, Any] | DimensionRecord, 

296 conform: bool = True, 

297 update: bool = False, 

298 ) -> bool | dict[str, Any]: 

299 raise NotImplementedError() 

300 

301 def queryDatasetTypes( 

302 self, 

303 expression: Any = ..., 

304 *, 

305 components: bool = False, 

306 missing: list[str] | None = None, 

307 ) -> Iterable[DatasetType]: 

308 query = convert_dataset_type_arg_to_glob_string_list(expression) 

309 request = QueryDatasetTypesRequestModel(search=query.search) 

310 response = self._connection.post("query_dataset_types", request) 

311 model = parse_model(response, QueryDatasetTypesResponseModel) 

312 if missing is not None: 

313 missing.extend(model.missing) 

314 

315 result = [] 

316 for dt in model.dataset_types: 

317 if dt.name in query.explicit_dataset_types: 

318 # Users are permitted to pass in already-existing DatasetType 

319 # instances, and we are supposed to preserve their overridden 

320 # storage class etc. 

321 result.append(query.explicit_dataset_types[dt.name]) 

322 else: 

323 result.append(DatasetType.from_simple(dt, self.dimensions)) 

324 

325 return result 

326 

327 def queryCollections( 

328 self, 

329 expression: Any = ..., 

330 datasetType: DatasetType | None = None, 

331 collectionTypes: Iterable[CollectionType] | CollectionType = CollectionType.all(), 

332 flattenChains: bool = False, 

333 includeChains: bool | None = None, 

334 ) -> Sequence[str]: 

335 return self._butler.collections.query( 

336 expression, 

337 collection_types=set(ensure_iterable(collectionTypes)), 

338 flatten_chains=flattenChains, 

339 include_chains=includeChains, 

340 ) 

341 

342 @property 

343 def storageClasses(self) -> StorageClassFactory: 

344 return self._butler.storageClasses 

345 

346 

347def _is_component_dataset_type(dataset_type: DatasetType | str) -> bool: 

348 """Return true if the given dataset type refers to a component.""" 

349 if isinstance(dataset_type, DatasetType): 

350 return dataset_type.component() is not None 

351 elif isinstance(dataset_type, str): 

352 parent, component = DatasetType.splitDatasetTypeName(dataset_type) 

353 return component is not None 

354 else: 

355 raise TypeError(f"Expected DatasetType or str, got {dataset_type!r}")