Coverage for python/lsst/daf/butler/_registry_shim.py: 59%

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

31 

32import contextlib 

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

34from typing import TYPE_CHECKING, Any 

35 

36from ._collection_type import CollectionType 

37from ._dataset_ref import DatasetId, DatasetIdGenEnum, DatasetRef 

38from ._dataset_type import DatasetType 

39from ._storage_class import StorageClassFactory 

40from ._timespan import Timespan 

41from .dimensions import ( 

42 DataCoordinate, 

43 DataId, 

44 DimensionElement, 

45 DimensionGroup, 

46 DimensionRecord, 

47 DimensionUniverse, 

48) 

49from .registry._collection_summary import CollectionSummary 

50from .registry._defaults import RegistryDefaults 

51from .registry._registry_base import RegistryBase 

52 

53if TYPE_CHECKING: 

54 from .direct_butler import DirectButler 

55 from .registry._registry import CollectionArgType 

56 from .registry.interfaces import ObsCoreTableManager 

57 

58 

59class RegistryShim(RegistryBase): 

60 """Implementation of `Registry` interface exposed to clients by `Butler`. 

61 

62 Parameters 

63 ---------- 

64 butler : `DirectButler` 

65 Data butler instance. 

66 

67 Notes 

68 ----- 

69 This shim implementation of `Registry` forwards all methods to an actual 

70 Registry instance which is internal to Butler or to Butler methods. Its 

71 purpose is to provide a stable interface to many client-visible operations 

72 while we perform re-structuring of Registry and Butler implementations. 

73 """ 

74 

75 def __init__(self, butler: DirectButler): 

76 super().__init__(butler) 

77 self._registry = butler._registry 

78 

79 def isWriteable(self) -> bool: 

80 # Docstring inherited from a base class. 

81 return self._registry.isWriteable() 

82 

83 @property 

84 def dimensions(self) -> DimensionUniverse: 

85 # Docstring inherited from a base class. 

86 return self._registry.dimensions 

87 

88 @property 

89 def defaults(self) -> RegistryDefaults: 

90 # Docstring inherited from a base class. 

91 return self._registry.defaults 

92 

93 @defaults.setter 

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

95 # Docstring inherited from a base class. 

96 self._registry.defaults = value 

97 

98 def refresh(self) -> None: 

99 # Docstring inherited from a base class. 

100 self._registry.refresh() 

101 

102 def refresh_collection_summaries(self) -> None: 

103 # Docstring inherited from a base class. 

104 self._registry.refresh_collection_summaries() 

105 

106 def caching_context(self) -> contextlib.AbstractContextManager[None]: 

107 # Docstring inherited from a base class. 

108 return self._butler._caching_context() 

109 

110 @contextlib.contextmanager 

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

112 # Docstring inherited from a base class. 

113 with self._registry.transaction(savepoint=savepoint): 

114 yield 

115 

116 def resetConnectionPool(self) -> None: 

117 # Docstring inherited from a base class. 

118 self._registry.resetConnectionPool() 

119 

120 def registerCollection( 

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

122 ) -> bool: 

123 # Docstring inherited from a base class. 

124 return self._registry.registerCollection(name, type, doc) 

125 

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

127 # Docstring inherited from a base class. 

128 return self._registry.getCollectionType(name) 

129 

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

131 # Docstring inherited from a base class. 

132 return self._registry.registerRun(name, doc) 

133 

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

135 # Docstring inherited from a base class. 

136 self._registry.removeCollection(name) 

137 

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

139 # Docstring inherited from a base class. 

140 return self._registry.getCollectionChain(parent) 

141 

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

143 # Docstring inherited from a base class. 

144 self._registry.setCollectionChain(parent, children, flatten=flatten) 

145 

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

147 # Docstring inherited from a base class. 

148 return self._registry.getCollectionParentChains(collection) 

149 

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

151 # Docstring inherited from a base class. 

152 return self._registry.getCollectionDocumentation(collection) 

153 

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

155 # Docstring inherited from a base class. 

156 self._registry.setCollectionDocumentation(collection, doc) 

157 

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

159 # Docstring inherited from a base class. 

160 return self._registry.getCollectionSummary(collection) 

161 

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

163 # Docstring inherited from a base class. 

164 return self._registry.registerDatasetType(datasetType) 

165 

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

167 # Docstring inherited from a base class. 

168 self._registry.removeDatasetType(name) 

169 

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

171 # Docstring inherited from a base class. 

172 return self._registry.getDatasetType(name) 

173 

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

175 # Docstring inherited from a base class. 

176 return self._registry.supportsIdGenerationMode(mode) 

177 

178 def findDataset( 

179 self, 

180 datasetType: DatasetType | str, 

181 dataId: DataId | None = None, 

182 *, 

183 collections: CollectionArgType | None = None, 

184 timespan: Timespan | None = None, 

185 **kwargs: Any, 

186 ) -> DatasetRef | None: 

187 # Docstring inherited from a base class. 

188 return self._registry.findDataset( 

189 datasetType, dataId, collections=collections, timespan=timespan, **kwargs 

190 ) 

191 

192 def insertDatasets( 

193 self, 

194 datasetType: DatasetType | str, 

195 dataIds: Iterable[DataId], 

196 run: str | None = None, 

197 expand: bool = True, 

198 idGenerationMode: DatasetIdGenEnum = DatasetIdGenEnum.UNIQUE, 

199 ) -> list[DatasetRef]: 

200 # Docstring inherited from a base class. 

201 return self._registry.insertDatasets(datasetType, dataIds, run, expand, idGenerationMode) 

202 

203 def _importDatasets(self, datasets: Iterable[DatasetRef], expand: bool = True) -> list[DatasetRef]: 

204 # Docstring inherited from a base class. 

205 return self._registry._importDatasets(datasets, expand) 

206 

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

208 # Docstring inherited from a base class. 

209 return self._registry.getDataset(id) 

210 

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

212 # Docstring inherited from a base class. 

213 self._registry.removeDatasets(refs) 

214 

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

216 # Docstring inherited from a base class. 

217 self._registry.associate(collection, refs) 

218 

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

220 # Docstring inherited from a base class. 

221 self._registry.disassociate(collection, refs) 

222 

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

224 # Docstring inherited from a base class. 

225 self._registry.certify(collection, refs, timespan) 

226 

227 def decertify( 

228 self, 

229 collection: str, 

230 datasetType: str | DatasetType, 

231 timespan: Timespan, 

232 *, 

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

234 ) -> None: 

235 # Docstring inherited from a base class. 

236 self._registry.decertify(collection, datasetType, timespan, dataIds=dataIds) 

237 

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

239 # Docstring inherited from a base class. 

240 return self._registry.getDatasetLocations(ref) 

241 

242 def expandDataId( 

243 self, 

244 dataId: DataId | None = None, 

245 *, 

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

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

248 withDefaults: bool = True, 

249 **kwargs: Any, 

250 ) -> DataCoordinate: 

251 # Docstring inherited from a base class. 

252 return self._registry.expandDataId( 

253 dataId, dimensions=dimensions, records=records, withDefaults=withDefaults, **kwargs 

254 ) 

255 

256 def insertDimensionData( 

257 self, 

258 element: DimensionElement | str, 

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

260 conform: bool = True, 

261 replace: bool = False, 

262 skip_existing: bool = False, 

263 ) -> None: 

264 # Docstring inherited from a base class. 

265 self._registry.insertDimensionData( 

266 element, *data, conform=conform, replace=replace, skip_existing=skip_existing 

267 ) 

268 

269 def syncDimensionData( 

270 self, 

271 element: DimensionElement | str, 

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

273 conform: bool = True, 

274 update: bool = False, 

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

276 # Docstring inherited from a base class. 

277 return self._registry.syncDimensionData(element, row, conform, update) 

278 

279 def queryDatasetTypes( 

280 self, 

281 expression: Any = ..., 

282 *, 

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

284 ) -> Iterable[DatasetType]: 

285 # Docstring inherited from a base class. 

286 return self._registry.queryDatasetTypes(expression, missing=missing) 

287 

288 def queryCollections( 

289 self, 

290 expression: Any = ..., 

291 datasetType: DatasetType | None = None, 

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

293 flattenChains: bool = False, 

294 includeChains: bool | None = None, 

295 ) -> Sequence[str]: 

296 # Docstring inherited from a base class. 

297 return self._registry.queryCollections( 

298 expression, datasetType, collectionTypes, flattenChains, includeChains 

299 ) 

300 

301 @property 

302 def obsCoreTableManager(self) -> ObsCoreTableManager | None: 

303 # Docstring inherited from a base class. 

304 return self._registry.obsCoreTableManager 

305 

306 @property 

307 def storageClasses(self) -> StorageClassFactory: 

308 return self._registry.storageClasses