Coverage for python/lsst/daf/butler/registry/interfaces/_bridge.py: 88%

72 statements  

« prev     ^ index     » next       coverage.py v7.17.2, created at 2026-11-06 09:46 +0000

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/>. 

27from __future__ import annotations 

28 

29__all__ = ("DatasetIdRef", "DatastoreRegistryBridge", "DatastoreRegistryBridgeManager", "FakeDatasetRef") 

30 

31from abc import ABC, abstractmethod 

32from collections.abc import Collection, Iterable 

33from contextlib import AbstractContextManager 

34from typing import TYPE_CHECKING, Any 

35 

36from lsst.utils.classes import immutable 

37 

38from ..._dataset_ref import DatasetId, DatasetRef 

39from ._versioning import VersionedExtension, VersionTuple 

40 

41if TYPE_CHECKING: 

42 from ..._dataset_ref import DatasetDatastoreRecords 

43 from ..._dataset_type import DatasetType 

44 from ...datastore import DatastoreTransaction 

45 from ...datastore.stored_file_info import StoredDatastoreItemInfo 

46 from ...dimensions import DimensionUniverse 

47 from ._database import Database, StaticTablesContext 

48 from ._datasets import DatasetRecordStorageManager 

49 from ._opaque import OpaqueTableStorage, OpaqueTableStorageManager 

50 

51 

52@immutable 

53class FakeDatasetRef: 

54 """A fake `DatasetRef` that can be used internally by butler where 

55 only the dataset ID is available. 

56 

57 Should only be used when registry can not be used to create a full 

58 `DatasetRef` from the ID. A particular use case is during dataset 

59 deletion when solely the ID is available. 

60 

61 Parameters 

62 ---------- 

63 id : `DatasetId` 

64 The dataset ID. 

65 """ 

66 

67 __slots__ = ("id",) 

68 

69 def __init__(self, id: DatasetId): 

70 self.id = id 

71 

72 def __str__(self) -> str: 

73 return f"dataset_id={self.id}" 

74 

75 def __repr__(self) -> str: 

76 return f"FakeDatasetRef({self.id})" 

77 

78 def __eq__(self, other: Any) -> bool: 

79 try: 

80 return self.id == other.id 

81 except AttributeError: 

82 return NotImplemented 

83 

84 def __hash__(self) -> int: 

85 return hash(self.id) 

86 

87 id: DatasetId 

88 """Unique identifier for this dataset. 

89 """ 

90 

91 @property 

92 def datasetType(self) -> DatasetType: 

93 raise AttributeError("A FakeDatasetRef can not be associated with a valid DatasetType") 

94 

95 @property 

96 def _datastore_records(self) -> DatasetDatastoreRecords | None: 

97 raise AttributeError("A FakeDatasetRef can not be associated with datastore records") 

98 

99 

100DatasetIdRef = DatasetRef | FakeDatasetRef 

101"""A type-annotation alias that matches both `DatasetRef` and `FakeDatasetRef`. 

102""" 

103 

104 

105class DatastoreRegistryBridge(ABC): 

106 """An abstract base class that defines the interface that a `Datastore` 

107 uses to communicate with a `Registry`. 

108 

109 Parameters 

110 ---------- 

111 datastoreName : `str` 

112 Name of the `Datastore` as it should appear in `Registry` tables 

113 referencing it. 

114 """ 

115 

116 def __init__(self, datastoreName: str): 

117 self.datastoreName = datastoreName 

118 

119 @abstractmethod 

120 def insert(self, refs: Iterable[DatasetIdRef]) -> None: 

121 """Record that a datastore holds the given datasets. 

122 

123 Parameters 

124 ---------- 

125 refs : `~collections.abc.Iterable` of `DatasetIdRef` 

126 References to the datasets. 

127 

128 Raises 

129 ------ 

130 ConflictingDefinitionError 

131 Raised if any of the ``refs`` is already registered. 

132 """ 

133 raise NotImplementedError() 

134 

135 @abstractmethod 

136 def ensure(self, refs: Iterable[DatasetIdRef]) -> None: 

137 """Record that a datastore holds the given datasets, skipping if 

138 the ref is already registered. 

139 

140 Parameters 

141 ---------- 

142 refs : `~collections.abc.Iterable` of `DatasetIdRef` 

143 References to the datasets. 

144 """ 

145 raise NotImplementedError() 

146 

147 @abstractmethod 

148 def forget(self, refs: Iterable[DatasetIdRef]) -> None: 

149 """Remove dataset location information without any attempt to put it 

150 in the trash while waiting for external deletes. 

151 

152 This should be used only to implement `Datastore.forget`, or in cases 

153 where deleting the actual datastore artifacts cannot fail. 

154 

155 Parameters 

156 ---------- 

157 refs : `~collections.abc.Iterable` of `DatasetIdRef` 

158 References to the datasets. 

159 """ 

160 raise NotImplementedError() 

161 

162 @abstractmethod 

163 def moveToTrash(self, refs: Iterable[DatasetIdRef], transaction: DatastoreTransaction | None) -> None: 

164 """Move dataset location information to trash. 

165 

166 Parameters 

167 ---------- 

168 refs : `~collections.abc.Iterable` of `DatasetIdRef` 

169 References to the datasets. 

170 transaction : `DatastoreTransaction` or `None` 

171 Transaction object. Can be `None` in some bridges or if no rollback 

172 is required. 

173 """ 

174 raise NotImplementedError() 

175 

176 @abstractmethod 

177 def check(self, datasets: Iterable[DatasetId]) -> set[DatasetId]: 

178 """Check which datasets are present in this datastore. 

179 

180 Parameters 

181 ---------- 

182 datasets : `~collections.abc.Iterable` of `DatasetId` 

183 UUIDs of the datasets to check. 

184 

185 Returns 

186 ------- 

187 present : `set` [ `DatasetId` ] 

188 Datasets from ``datasets`` that are recorded as being in this 

189 datastore. 

190 """ 

191 raise NotImplementedError() 

192 

193 @abstractmethod 

194 def emptyTrash( 

195 self, 

196 records_table: OpaqueTableStorage | None = None, 

197 record_class: type[StoredDatastoreItemInfo] | None = None, 

198 record_column: str | None = None, 

199 selected_ids: Collection[DatasetId] | None = None, 

200 dry_run: bool = False, 

201 ) -> AbstractContextManager[ 

202 tuple[Iterable[tuple[DatasetIdRef, StoredDatastoreItemInfo | None]], set[str] | None] 

203 ]: 

204 """Retrieve all the dataset ref IDs that are in the trash 

205 associated for this datastore, and then remove them if the context 

206 exists without an exception being raised. 

207 

208 Parameters 

209 ---------- 

210 records_table : `OpaqueTableStorage`, optional 

211 Table of records to query with the trash records. 

212 record_class : `type` of `StoredDatastoreItemInfo`, optional 

213 Class to use when reading records from ``records_table``. 

214 record_column : `str`, optional 

215 Name of the column in records_table that refers to the artifact. 

216 selected_ids : `collections.abc.Collection` [ `DatasetId` ] \ 

217 or `None`, optional 

218 If provided, collection of IDs that should be trashed. Only records 

219 within this selection will be yielded and then removed. This 

220 can be used to allow a subset of the trash table to be emptied. 

221 If an empty set is given no artifacts will be trashed. If `None` 

222 the full list from the trash table will be used. 

223 dry_run : `bool`, optional 

224 If `True`, the trash table will be queried and results reported 

225 but no artifacts will be removed. 

226 

227 Yields 

228 ------ 

229 matches : `~collections.abc.Iterable` [`DatasetIdRef`, \ 

230 `StoredDatastoreItemInfo`] 

231 The IDs of datasets that can be safely removed from this datastore 

232 and the corresponding information from the records table. 

233 Can be empty. 

234 artifacts_to_keep : `set` of `str`, optional 

235 Any external artifacts that are known to the table but which should 

236 not be deleted. If `None`, the caller should check themselves. 

237 

238 Examples 

239 -------- 

240 Typical usage by a Datastore is something like:: 

241 

242 with self.bridge.emptyTrash() as trashed: 

243 iter, to_keep = trashed 

244 for ref, info in iter: 

245 # Remove artifacts associated with id, 

246 # raise an exception if something goes wrong. 

247 

248 Notes 

249 ----- 

250 The object yielded by the context manager may be a single-pass 

251 iterator. If multiple passes are required, it should be converted to 

252 a `list` or other container. 

253 

254 Datastores should never raise (except perhaps in testing) when an 

255 artifact cannot be removed only because it is already gone - this 

256 condition is an unavoidable outcome of concurrent delete operations, 

257 and must not be considered and error for those to be safe. 

258 

259 If a table record is provided the trashed records will be deleted 

260 when the context manager completes. 

261 """ 

262 raise NotImplementedError() 

263 

264 datastoreName: str 

265 """The name of the `Datastore` as it should appear in `Registry` tables 

266 (`str`). 

267 """ 

268 

269 

270class DatastoreRegistryBridgeManager(VersionedExtension): 

271 """An abstract base class that defines the interface between `Registry` 

272 and `Datastore` when a new `Datastore` is constructed. 

273 

274 Parameters 

275 ---------- 

276 opaque : `OpaqueTableStorageManager` 

277 Manager object for opaque table storage in the `Registry`. 

278 universe : `DimensionUniverse` 

279 All dimensions know to the `Registry`. 

280 registry_schema_version : `VersionTuple` or `None`, optional 

281 Version of registry schema. 

282 

283 Notes 

284 ----- 

285 Datastores are passed an instance of `DatastoreRegistryBridgeManager` at 

286 construction, and should use it to obtain and keep any of the following: 

287 

288 - a `DatastoreRegistryBridge` instance to record in the `Registry` what is 

289 present in the datastore (needed by all datastores that are not just 

290 forwarders); 

291 

292 - one or more `OpaqueTableStorage` instance if they wish to store internal 

293 records in the `Registry` database; 

294 

295 - the `DimensionUniverse`, if they need it to (e.g.) construct or validate 

296 filename templates. 

297 """ 

298 

299 def __init__( 

300 self, 

301 *, 

302 opaque: OpaqueTableStorageManager, 

303 universe: DimensionUniverse, 

304 registry_schema_version: VersionTuple | None = None, 

305 ): 

306 super().__init__(registry_schema_version=registry_schema_version) 

307 self.opaque = opaque 

308 self.universe = universe 

309 

310 @abstractmethod 

311 def clone( 

312 self, 

313 *, 

314 db: Database, 

315 opaque: OpaqueTableStorageManager, 

316 ) -> DatastoreRegistryBridgeManager: 

317 """Make an independent copy of this manager instance bound to new 

318 instances of `Database` and other managers. 

319 

320 Parameters 

321 ---------- 

322 db : `Database` 

323 New `Database` object to use when instantiating the manager. 

324 opaque : `OpaqueTableStorageManager` 

325 New `OpaqueTableStorageManager` object to use when instantiating 

326 the manager. 

327 

328 Returns 

329 ------- 

330 instance : `DatastoreRegistryBridgeManager` 

331 New manager instance with the same configuration as this instance, 

332 but bound to a new Database object. 

333 """ 

334 raise NotImplementedError() 

335 

336 @classmethod 

337 @abstractmethod 

338 def initialize( 

339 cls, 

340 db: Database, 

341 context: StaticTablesContext, 

342 *, 

343 opaque: OpaqueTableStorageManager, 

344 datasets: type[DatasetRecordStorageManager], 

345 universe: DimensionUniverse, 

346 registry_schema_version: VersionTuple | None = None, 

347 ) -> DatastoreRegistryBridgeManager: 

348 """Construct an instance of the manager. 

349 

350 Parameters 

351 ---------- 

352 db : `Database` 

353 Interface to the underlying database engine and namespace. 

354 context : `StaticTablesContext` 

355 Context object obtained from `Database.declareStaticTables`; used 

356 to declare any tables that should always be present in a layer 

357 implemented with this manager. 

358 opaque : `OpaqueTableStorageManager` 

359 Registry manager object for opaque (to Registry) tables, provided 

360 to allow Datastores to store their internal information inside the 

361 Registry database. 

362 datasets : subclass of `DatasetRecordStorageManager` 

363 Concrete class that will be used to manage the core dataset tables 

364 in this registry; should be used only to create foreign keys to 

365 those tables. 

366 universe : `DimensionUniverse` 

367 All dimensions known to the registry. 

368 registry_schema_version : `VersionTuple` or `None` 

369 Schema version of this extension as defined in registry. 

370 

371 Returns 

372 ------- 

373 manager : `DatastoreRegistryBridgeManager` 

374 An instance of a concrete `DatastoreRegistryBridgeManager` 

375 subclass. 

376 """ 

377 raise NotImplementedError() 

378 

379 @abstractmethod 

380 def refresh(self) -> None: 

381 """Ensure all other operations on this manager are aware of any 

382 collections that may have been registered by other clients since it 

383 was initialized or last refreshed. 

384 """ 

385 raise NotImplementedError() 

386 

387 @abstractmethod 

388 def register(self, name: str, *, ephemeral: bool = True) -> DatastoreRegistryBridge: 

389 """Register a new `Datastore` associated with this `Registry`. 

390 

391 This method should be called by all `Datastore` classes aside from 

392 those that only forward storage to other datastores. 

393 

394 Parameters 

395 ---------- 

396 name : `str` 

397 Name of the datastore, as it should appear in `Registry` tables. 

398 ephemeral : `bool`, optional 

399 If `True` (`False` is default), return a bridge object that is 

400 backed by storage that will not last past the end of the current 

401 process. This should be used whenever the same is true of the 

402 dataset's artifacts. 

403 

404 Returns 

405 ------- 

406 bridge : `DatastoreRegistryBridge` 

407 Object that provides the interface this `Datastore` should use to 

408 communicate with the `Registry`. 

409 """ 

410 raise NotImplementedError() 

411 

412 @abstractmethod 

413 def findDatastores(self, ref: DatasetIdRef) -> Iterable[str]: 

414 """Retrieve datastore locations for a given dataset. 

415 

416 Parameters 

417 ---------- 

418 ref : `DatasetIdRef` 

419 A reference to the dataset for which to retrieve storage 

420 information. 

421 

422 Returns 

423 ------- 

424 datastores : `~collections.abc.Iterable` [ `str` ] 

425 All the matching datastores holding this dataset. Empty if the 

426 dataset does not exist anywhere. 

427 

428 Raises 

429 ------ 

430 AmbiguousDatasetError 

431 Raised if ``ref.id`` is `None`. 

432 """ 

433 raise NotImplementedError() 

434 

435 opaque: OpaqueTableStorageManager 

436 """Registry manager object for opaque (to Registry) tables, provided 

437 to allow Datastores to store their internal information inside the 

438 Registry database. 

439 """ 

440 

441 universe: DimensionUniverse 

442 """All dimensions known to the `Registry`. 

443 """