Coverage for python/lsst/daf/butler/datastores/inMemoryDatastore.py: 91%

217 statements  

« prev     ^ index     » next       coverage.py v7.16.2, created at 2026-11-06 01:43 -0800

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 

28"""In-memory datastore.""" 

29 

30from __future__ import annotations 

31 

32__all__ = ("InMemoryDatastore", "StoredMemoryItemInfo") 

33 

34import logging 

35import time 

36from collections.abc import Collection, Iterable, Mapping 

37from dataclasses import dataclass 

38from typing import TYPE_CHECKING, Any 

39from urllib.parse import urlencode 

40 

41from lsst.daf.butler import DatasetId, DatasetRef, StorageClass 

42from lsst.daf.butler._exceptions import DatasetTypeNotSupportedError 

43from lsst.daf.butler.datastore import DatasetRefURIs, DatastoreConfig, DuplicateIdError 

44from lsst.daf.butler.datastore.generic_base import GenericBaseDatastore, post_process_get 

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

46from lsst.daf.butler.datastore.stored_file_info import StoredDatastoreItemInfo 

47from lsst.daf.butler.utils import transactional 

48from lsst.resources import ResourcePath, ResourcePathExpression 

49 

50if TYPE_CHECKING: 

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

52 from lsst.daf.butler.datastore import DatastoreOpaqueTable 

53 from lsst.daf.butler.datastores.file_datastore.retrieve_artifacts import ArtifactIndexInfo 

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

55 

56log = logging.getLogger(__name__) 

57 

58 

59@dataclass(frozen=True, slots=True) 

60class StoredMemoryItemInfo(StoredDatastoreItemInfo): 

61 """Internal InMemoryDatastore Metadata associated with a stored 

62 DatasetRef. 

63 """ 

64 

65 timestamp: float 

66 """Unix timestamp indicating the time the dataset was stored.""" 

67 

68 storageClass: StorageClass 

69 """StorageClass associated with the dataset.""" 

70 

71 parentID: DatasetId 

72 """ID of the parent `DatasetRef` if this entry is a concrete 

73 composite. Not used if the dataset being stored is not a 

74 virtual component of a composite 

75 """ 

76 

77 

78class InMemoryDatastore(GenericBaseDatastore[StoredMemoryItemInfo]): 

79 """Basic Datastore for writing to an in memory cache. 

80 

81 This datastore is ephemeral in that the contents of the datastore 

82 disappear when the Python process completes. This also means that 

83 other processes can not access this datastore. 

84 

85 Parameters 

86 ---------- 

87 config : `DatastoreConfig` or `str` 

88 Configuration. 

89 bridgeManager : `DatastoreRegistryBridgeManager` 

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

91 

92 Notes 

93 ----- 

94 InMemoryDatastore does not support any file-based ingest. 

95 """ 

96 

97 defaultConfigFile = "datastores/inMemoryDatastore.yaml" 

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

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

100 """ 

101 

102 isEphemeral = True 

103 """A new datastore is created every time and datasets disappear when 

104 the process shuts down.""" 

105 

106 datasets: dict[DatasetId, Any] 

107 """Internal storage of datasets indexed by dataset ID.""" 

108 

109 records: dict[DatasetId, StoredMemoryItemInfo] 

110 """Internal records about stored datasets.""" 

111 

112 def __init__( 

113 self, 

114 config: DatastoreConfig, 

115 bridgeManager: DatastoreRegistryBridgeManager, 

116 ): 

117 super().__init__(config, bridgeManager) 

118 

119 # Name ourselves with the timestamp the datastore 

120 # was created. 

121 self.name = f"{type(self).__name__}@{time.time()}" 

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

123 

124 # Storage of datasets, keyed by dataset_id 

125 self.datasets: dict[DatasetId, Any] = {} 

126 

127 # Records is distinct in order to track concrete composite components 

128 # where we register multiple components for a single dataset. 

129 self.records: dict[DatasetId, StoredMemoryItemInfo] = {} 

130 

131 # Related records that share the same parent 

132 self.related: dict[DatasetId, set[DatasetId]] = {} 

133 

134 self._trashedIds: set[DatasetId] = set() 

135 

136 @classmethod 

137 def _create_from_config( 

138 cls, 

139 config: DatastoreConfig, 

140 bridgeManager: DatastoreRegistryBridgeManager, 

141 butlerRoot: ResourcePathExpression | None, 

142 ) -> InMemoryDatastore: 

143 return InMemoryDatastore(config, bridgeManager) 

144 

145 def clone(self, bridgeManager: DatastoreRegistryBridgeManager) -> InMemoryDatastore: 

146 clone = InMemoryDatastore(self.config, bridgeManager) 

147 # Sharing these objects is not thread-safe, but this class is only used 

148 # in single-threaded test code. 

149 clone.datasets = self.datasets 

150 clone.records = self.records 

151 clone.related = self.related 

152 clone._trashedIds = self._trashedIds 

153 return clone 

154 

155 @classmethod 

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

157 """Set any filesystem-dependent config options for this Datastore to 

158 be appropriate for a new empty repository with the given root. 

159 

160 Does nothing in this implementation. 

161 

162 Parameters 

163 ---------- 

164 root : `str` 

165 Filesystem path to the root of the data repository. 

166 config : `Config` 

167 A `Config` to update. Only the subset understood by 

168 this component will be updated. Will not expand 

169 defaults. 

170 full : `Config` 

171 A complete config with all defaults expanded that can be 

172 converted to a `DatastoreConfig`. Read-only and will not be 

173 modified by this method. 

174 Repository-specific options that should not be obtained 

175 from defaults when Butler instances are constructed 

176 should be copied from ``full`` to ``config``. 

177 overwrite : `bool`, optional 

178 If `False`, do not modify a value in ``config`` if the value 

179 already exists. Default is always to overwrite with the provided 

180 ``root``. 

181 

182 Notes 

183 ----- 

184 If a keyword is explicitly defined in the supplied ``config`` it 

185 will not be overridden by this method if ``overwrite`` is `False`. 

186 This allows explicit values set in external configs to be retained. 

187 """ 

188 return 

189 

190 def _get_stored_item_info(self, dataset_id: DatasetId) -> StoredMemoryItemInfo: 

191 # Docstring inherited from GenericBaseDatastore. 

192 return self.records[dataset_id] 

193 

194 def _remove_stored_item_info(self, dataset_id: DatasetId) -> None: 

195 # Docstring inherited from GenericBaseDatastore. 

196 # If a component has been removed previously then we can sometimes 

197 # be asked to remove it again. Other datastores ignore this 

198 # so also ignore here 

199 if dataset_id not in self.records: 

200 return 

201 record = self.records[dataset_id] 

202 del self.records[dataset_id] 

203 self.related[record.parentID].remove(dataset_id) 

204 

205 def removeStoredItemInfo(self, ref: DatasetIdRef) -> None: 

206 """Remove information about the file associated with this dataset. 

207 

208 Parameters 

209 ---------- 

210 ref : `DatasetRef` 

211 The dataset that has been removed. 

212 

213 Notes 

214 ----- 

215 This method is actually not used by this implementation, but there are 

216 some tests that check that this method works, so we keep it for now. 

217 """ 

218 self._remove_stored_item_info(ref.id) 

219 

220 def _get_dataset_info(self, dataset_id: DatasetId) -> tuple[DatasetId, StoredMemoryItemInfo]: 

221 """Check that the dataset is present and return the real ID and 

222 associated information. 

223 

224 Parameters 

225 ---------- 

226 dataset_id : `DatasetRef` 

227 Target `DatasetRef` 

228 

229 Returns 

230 ------- 

231 realID : `int` 

232 The dataset ID associated with this ref that should be used. This 

233 could either be the ID of the supplied `DatasetRef` or the parent. 

234 storageInfo : `StoredMemoryItemInfo` 

235 Associated storage information. 

236 

237 Raises 

238 ------ 

239 FileNotFoundError 

240 Raised if the dataset is not present in this datastore. 

241 """ 

242 try: 

243 storedItemInfo = self._get_stored_item_info(dataset_id) 

244 except KeyError: 

245 raise FileNotFoundError(f"No such file dataset in memory: {dataset_id}") from None 

246 realID = dataset_id 

247 if storedItemInfo.parentID is not None: 247 ↛ 250line 247 didn't jump to line 250 because the condition on line 247 was always true

248 realID = storedItemInfo.parentID 

249 

250 if realID not in self.datasets: 250 ↛ 251line 250 didn't jump to line 251 because the condition on line 250 was never true

251 raise FileNotFoundError(f"No such file dataset in memory: {dataset_id}") 

252 

253 return realID, storedItemInfo 

254 

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

256 """Check if the dataset is known to the datastore. 

257 

258 This datastore does not distinguish dataset existence from knowledge 

259 of a dataset. 

260 

261 Parameters 

262 ---------- 

263 ref : `DatasetRef` 

264 Reference to the required dataset. 

265 

266 Returns 

267 ------- 

268 exists : `bool` 

269 `True` if the dataset is known to the datastore. 

270 """ 

271 return self.exists(ref) 

272 

273 def exists(self, ref: DatasetRef) -> bool: 

274 """Check if the dataset exists in the datastore. 

275 

276 Parameters 

277 ---------- 

278 ref : `DatasetRef` 

279 Reference to the required dataset. 

280 

281 Returns 

282 ------- 

283 exists : `bool` 

284 `True` if the entity exists in the `Datastore`. 

285 """ 

286 try: 

287 self._get_dataset_info(ref.id) 

288 except FileNotFoundError: 

289 return False 

290 return True 

291 

292 def get( 

293 self, 

294 ref: DatasetRef, 

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

296 storageClass: StorageClass | str | None = None, 

297 ) -> Any: 

298 """Load an InMemoryDataset from the store. 

299 

300 Parameters 

301 ---------- 

302 ref : `DatasetRef` 

303 Reference to the required Dataset. 

304 parameters : `dict` 

305 `StorageClass`-specific parameters that specify, for example, 

306 a slice of the dataset to be loaded. 

307 storageClass : `StorageClass` or `str`, optional 

308 The storage class to be used to override the Python type 

309 returned by this method. By default the returned type matches 

310 the dataset type definition for this dataset. Specifying a 

311 read `StorageClass` can force a different type to be returned. 

312 This type must be compatible with the original type. 

313 

314 Returns 

315 ------- 

316 inMemoryDataset : `object` 

317 Requested dataset or slice thereof as an InMemoryDataset. 

318 

319 Raises 

320 ------ 

321 FileNotFoundError 

322 Requested dataset can not be retrieved. 

323 TypeError 

324 Return value from formatter has unexpected type. 

325 ValueError 

326 Formatter failed to process the dataset. 

327 """ 

328 log.debug("Retrieve %s from %s with parameters %s", ref, self.name, parameters) 

329 

330 realID, storedItemInfo = self._get_dataset_info(ref.id) 

331 

332 # We have a write storage class and a read storage class and they 

333 # can be different for concrete composites or if overridden. 

334 if storageClass is not None: 

335 ref = ref.overrideStorageClass(storageClass) 

336 refStorageClass = ref.datasetType.storageClass 

337 writeStorageClass = storedItemInfo.storageClass 

338 

339 component = ref.datasetType.component() 

340 

341 # Check that the supplied parameters are suitable for the type read 

342 # If this is a derived component we validate against the composite 

343 isDerivedComponent = False 

344 if component in writeStorageClass.derivedComponents: 

345 writeStorageClass.validateParameters(parameters) 

346 isDerivedComponent = True 

347 else: 

348 refStorageClass.validateParameters(parameters) 

349 

350 inMemoryDataset = self.datasets[realID] 

351 

352 # if this is a read only component we need to apply parameters 

353 # before we retrieve the component. We assume that the parameters 

354 # will affect the data globally, before the derived component 

355 # is selected. 

356 if isDerivedComponent: 

357 inMemoryDataset = writeStorageClass.delegate().handleParameters(inMemoryDataset, parameters) 

358 # Then disable parameters for later 

359 parameters = {} 

360 

361 # Check if we have a component. 

362 if component: 

363 # In-memory datastore must have stored the dataset as a single 

364 # object in the write storage class. We therefore use that 

365 # storage class delegate to obtain the component. 

366 compositeStorageClass = writeStorageClass 

367 if component not in writeStorageClass.allComponents(): 

368 # The component is defined by the storage class the caller is 

369 # converting to rather than the one used to write, so the 

370 # composite has to be converted before the component can be 

371 # extracted from it. 

372 parentStorageClass = ref.datasetType.parentStorageClass 

373 if parentStorageClass is not None: 373 ↛ 376line 373 didn't jump to line 376 because the condition on line 373 was always true

374 compositeStorageClass = parentStorageClass 

375 inMemoryDataset = parentStorageClass.coerce_type(inMemoryDataset) 

376 inMemoryDataset = compositeStorageClass.delegate().getComponent(inMemoryDataset, component) 

377 

378 # Since there is no formatter to process parameters, they all must be 

379 # passed to the assembler. 

380 inMemoryDataset = post_process_get( 

381 inMemoryDataset, refStorageClass, parameters, isComponent=component is not None 

382 ) 

383 

384 # Last minute type conversion. 

385 return refStorageClass.coerce_type(inMemoryDataset) 

386 

387 def put(self, inMemoryDataset: Any, ref: DatasetRef, provenance: DatasetProvenance | None = None) -> None: 

388 """Write a InMemoryDataset with a given `DatasetRef` to the store. 

389 

390 Parameters 

391 ---------- 

392 inMemoryDataset : `object` 

393 The dataset to store. 

394 ref : `DatasetRef` 

395 Reference to the associated Dataset. 

396 provenance : `DatasetProvenance` or `None`, optional 

397 Any provenance that should be attached to the serialized dataset. 

398 

399 Raises 

400 ------ 

401 TypeError 

402 Supplied object and storage class are inconsistent. 

403 DatasetTypeNotSupportedError 

404 The associated `DatasetType` is not handled by this datastore. 

405 

406 Notes 

407 ----- 

408 If the datastore is configured to reject certain dataset types it 

409 is possible that the put will fail and raise a 

410 `DatasetTypeNotSupportedError`. The main use case for this is to 

411 allow `ChainedDatastore` to put to multiple datastores without 

412 requiring that every datastore accepts the dataset. 

413 """ 

414 if not self.constraints.isAcceptable(ref): 

415 # Raise rather than use boolean return value. 

416 raise DatasetTypeNotSupportedError( 

417 f"Dataset {ref} has been rejected by this datastore via configuration." 

418 ) 

419 

420 # Do not override existing datasets. 

421 if ref.id in self.records: 

422 raise DuplicateIdError(f"Attempt to override existing dataset: {ref.id}") 

423 

424 # May need to coerce the in memory dataset to the correct 

425 # python type, otherwise parameters may not work. 

426 try: 

427 delegate = ref.datasetType.storageClass.delegate() 

428 except TypeError: 

429 # TypeError is raised when a storage class doesn't have a delegate. 

430 delegate = None 

431 if not delegate or not delegate.can_accept(inMemoryDataset): 

432 inMemoryDataset = ref.datasetType.storageClass.coerce_type(inMemoryDataset) 

433 

434 # Update provenance. 

435 if delegate: 435 ↛ 438line 435 didn't jump to line 438 because the condition on line 435 was always true

436 inMemoryDataset = delegate.add_provenance(inMemoryDataset, ref, provenance=provenance) 

437 

438 self.datasets[ref.id] = inMemoryDataset 

439 log.debug("Store %s in %s", ref, self.name) 

440 

441 # Store time we received this content, to allow us to optionally 

442 # expire it. Instead of storing a filename here, we include the 

443 # ID of this datasetRef so we can find it from components. 

444 itemInfo = StoredMemoryItemInfo(time.time(), ref.datasetType.storageClass, parentID=ref.id) 

445 

446 # We have to register this content with registry. 

447 # Currently this assumes we have a file so we need to use stub entries 

448 self.records[ref.id] = itemInfo 

449 self.related.setdefault(itemInfo.parentID, set()).add(ref.id) 

450 

451 if self._transaction is not None: 

452 self._transaction.registerUndo("put", self.remove, ref) 

453 

454 def getURIs(self, ref: DatasetRef, predict: bool = False) -> DatasetRefURIs: 

455 """Return URIs associated with dataset. 

456 

457 Parameters 

458 ---------- 

459 ref : `DatasetRef` 

460 Reference to the required dataset. 

461 predict : `bool`, optional 

462 If the datastore does not know about the dataset, controls whether 

463 it should return a predicted URI or not. 

464 

465 Returns 

466 ------- 

467 uris : `DatasetRefURIs` 

468 The URI to the primary artifact associated with this dataset (if 

469 the dataset was disassembled within the datastore this may be 

470 `None`), and the URIs to any components associated with the dataset 

471 artifact. (can be empty if there are no components). 

472 

473 Notes 

474 ----- 

475 The URIs returned for in-memory datastores are not usable but 

476 provide an indication of the associated dataset. 

477 """ 

478 # Include the dataID as a URI query 

479 query = urlencode(ref.dataId.required) 

480 

481 # if this has never been written then we have to guess 

482 if not self.exists(ref): 

483 if not predict: 

484 raise FileNotFoundError(f"Dataset {ref} not in this datastore") 

485 name = f"{ref.datasetType.name}" 

486 fragment = "#predicted" 

487 else: 

488 realID, _ = self._get_dataset_info(ref.id) 

489 name = f"{id(self.datasets[realID])}_{ref.id}" 

490 fragment = "" 

491 

492 return DatasetRefURIs(ResourcePath(f"mem://{name}?{query}{fragment}"), {}) 

493 

494 def getURI(self, ref: DatasetRef, predict: bool = False) -> ResourcePath: 

495 """URI to the Dataset. 

496 

497 Always uses "mem://" URI prefix. 

498 

499 Parameters 

500 ---------- 

501 ref : `DatasetRef` 

502 Reference to the required Dataset. 

503 predict : `bool` 

504 If `True`, allow URIs to be returned of datasets that have not 

505 been written. 

506 

507 Returns 

508 ------- 

509 uri : `str` 

510 URI pointing to the dataset within the datastore. If the 

511 dataset does not exist in the datastore, and if ``predict`` is 

512 `True`, the URI will be a prediction and will include a URI 

513 fragment "#predicted". 

514 If the datastore does not have entities that relate well 

515 to the concept of a URI the returned URI string will be 

516 descriptive. The returned URI is not guaranteed to be obtainable. 

517 

518 Raises 

519 ------ 

520 FileNotFoundError 

521 A URI has been requested for a dataset that does not exist and 

522 guessing is not allowed. 

523 AssertionError 

524 Raised if an internal error occurs. 

525 """ 

526 primary, _ = self.getURIs(ref, predict) 

527 if primary is None: 

528 # This should be impossible since this datastore does 

529 # not disassemble. This check also helps mypy. 

530 raise AssertionError(f"Unexpectedly got no URI for in-memory datastore for {ref}") 

531 return primary 

532 

533 def ingest_zip(self, zip_path: ResourcePath, transfer: str | None, *, dry_run: bool = False) -> None: 

534 raise NotImplementedError("Can only ingest a Zip into a file datastore.") 

535 

536 def retrieveArtifacts( 

537 self, 

538 refs: Iterable[DatasetRef], 

539 destination: ResourcePath, 

540 transfer: str = "auto", 

541 preserve_path: bool = True, 

542 overwrite: bool | None = False, 

543 write_index: bool = True, 

544 add_prefix: bool = False, 

545 ) -> dict[ResourcePath, ArtifactIndexInfo]: 

546 """Retrieve the file artifacts associated with the supplied refs. 

547 

548 Parameters 

549 ---------- 

550 refs : `~collections.abc.Iterable` of `DatasetRef` 

551 The datasets for which artifacts are to be retrieved. 

552 A single ref can result in multiple artifacts. The refs must 

553 be resolved. 

554 destination : `lsst.resources.ResourcePath` 

555 Location to write the artifacts. 

556 transfer : `str`, optional 

557 Method to use to transfer the artifacts. Must be one of the options 

558 supported by `lsst.resources.ResourcePath.transfer_from`. 

559 "move" is not allowed. 

560 preserve_path : `bool`, optional 

561 If `True` the full path of the artifact within the datastore 

562 is preserved. If `False` the final file component of the path 

563 is used. 

564 overwrite : `bool`, optional 

565 If `True` allow transfers to overwrite existing files at the 

566 destination. 

567 write_index : `bool`, optional 

568 If `True` write a file at the top level containing a serialization 

569 of a `ZipIndex` for the downloaded datasets. 

570 add_prefix : `bool`, optional 

571 If `True` and if ``preserve_path`` is `False`, apply a prefix to 

572 the filenames corresponding to some part of the dataset ref ID. 

573 This can be used to guarantee uniqueness. 

574 

575 Notes 

576 ----- 

577 Not implemented by this datastore. 

578 """ 

579 # Could conceivably launch a FileDatastore to use formatters to write 

580 # the data but this is fraught with problems. 

581 raise NotImplementedError("Can not write artifacts to disk from in-memory datastore.") 

582 

583 def forget(self, refs: Iterable[DatasetRef]) -> None: 

584 # Docstring inherited. 

585 refs = list(refs) 

586 for ref in refs: 

587 self._remove_stored_item_info(ref.id) 

588 

589 @transactional 

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

591 """Indicate to the Datastore that a dataset can be removed. 

592 

593 Parameters 

594 ---------- 

595 ref : `DatasetRef` or iterable thereof 

596 Reference to the required Dataset(s). 

597 ignore_errors : `bool`, optional 

598 Indicate that errors should be ignored. 

599 

600 Returns 

601 ------- 

602 None 

603 

604 Raises 

605 ------ 

606 FileNotFoundError 

607 Attempt to remove a dataset that does not exist. Only relevant 

608 if a single dataset ref is given. 

609 

610 Notes 

611 ----- 

612 Concurrency should not normally be an issue for the in memory datastore 

613 since all internal changes are isolated to solely this process and 

614 the registry only changes rows associated with this process. 

615 """ 

616 if isinstance(ref, DatasetRef): 

617 # Check that this dataset is known to datastore 

618 try: 

619 self._get_dataset_info(ref.id) 

620 except Exception as e: 

621 if ignore_errors: 621 ↛ 622line 621 didn't jump to line 622 because the condition on line 621 was never true

622 log.warning( 

623 "Error encountered moving dataset %s to trash in datastore %s: %s", ref, self.name, e 

624 ) 

625 else: 

626 raise 

627 log.debug("Trash %s in datastore %s", ref, self.name) 

628 ref_list = [ref] 

629 else: 

630 ref_list = list(ref) 

631 log.debug("Bulk trashing of datasets in datastore %s", self.name) 

632 

633 def _rollbackMoveToTrash(refs: Iterable[DatasetIdRef]) -> None: 

634 for ref in refs: 634 ↛ exitline 634 didn't return from function '_rollbackMoveToTrash' because the loop on line 634 didn't complete

635 self._trashedIds.remove(ref.id) 

636 

637 assert self._transaction is not None, "Must be in transaction" 

638 with self._transaction.undoWith(f"Trash {len(ref_list)} datasets", _rollbackMoveToTrash, ref_list): 

639 self._trashedIds.update(ref.id for ref in ref_list) 

640 

641 def emptyTrash( 

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

643 ) -> set[ResourcePath]: 

644 """Remove all datasets from the trash. 

645 

646 Parameters 

647 ---------- 

648 ignore_errors : `bool`, optional 

649 Ignore errors. 

650 refs : `collections.abc.Collection` [ `DatasetRef` ] or `None` 

651 Explicit list of datasets that can be removed from trash. If listed 

652 datasets are not already stored in the trash table they will be 

653 ignored. If `None` every entry in the trash table will be 

654 processed. 

655 dry_run : `bool`, optional 

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

657 but no artifacts will be removed. 

658 

659 Returns 

660 ------- 

661 removed : `set` [ `lsst.resources.ResourcePath` ] 

662 List of artifacts that were removed. Empty for this datastore. 

663 

664 Notes 

665 ----- 

666 The internal tracking of datasets is affected by this method and 

667 transaction handling is not supported if there is a problem before 

668 the datasets themselves are deleted. 

669 

670 Concurrency should not normally be an issue for the in memory datastore 

671 since all internal changes are isolated to solely this process and 

672 the registry only changes rows associated with this process. 

673 """ 

674 log.debug("Emptying trash in datastore %s", self.name) 

675 

676 trashed_ids = self._trashedIds 

677 if refs: 677 ↛ 682line 677 didn't jump to line 682 because the condition on line 677 was always true

678 selected_ids = {ref.id for ref in refs} 

679 if selected_ids is not None: 679 ↛ 682line 679 didn't jump to line 682 because the condition on line 679 was always true

680 trashed_ids = {tid for tid in trashed_ids if tid in selected_ids} 

681 

682 if dry_run: 682 ↛ 683line 682 didn't jump to line 683 because the condition on line 682 was never true

683 log.info( 

684 "Would attempt remove %s dataset%s.", len(trashed_ids), "s" if len(trashed_ids) != 1 else "" 

685 ) 

686 return set() 

687 

688 for dataset_id in trashed_ids: 

689 try: 

690 realID, _ = self._get_dataset_info(dataset_id) 

691 except FileNotFoundError: 

692 # Dataset already removed so ignore it 

693 continue 

694 except Exception as e: 

695 if ignore_errors: 

696 log.warning( 

697 "Emptying trash in datastore %s but encountered an error with dataset %s: %s", 

698 self.name, 

699 dataset_id, 

700 e, 

701 ) 

702 continue 

703 else: 

704 raise 

705 

706 # Determine whether all references to this dataset have been 

707 # removed and we can delete the dataset itself 

708 allRefs = self.related[realID] 

709 remainingRefs = allRefs - {dataset_id} 

710 if not remainingRefs: 710 ↛ 715line 710 didn't jump to line 715 because the condition on line 710 was always true

711 log.debug("Removing artifact %s from datastore %s", realID, self.name) 

712 del self.datasets[realID] 

713 

714 # Remove this entry 

715 self._remove_stored_item_info(dataset_id) 

716 

717 # Empty the trash table 

718 self._trashedIds = self._trashedIds - trashed_ids 

719 return set() 

720 

721 def validateConfiguration( 

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

723 ) -> None: 

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

725 

726 Parameters 

727 ---------- 

728 entities : `~collections.abc.Iterable` [`DatasetRef` | `DatasetType`\ 

729 | `StorageClass`] 

730 Entities to test against this configuration. Can be differing 

731 types. 

732 logFailures : `bool`, optional 

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

734 detected. 

735 

736 Returns 

737 ------- 

738 None 

739 

740 Raises 

741 ------ 

742 DatastoreValidationError 

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

744 All the problems are reported in a single exception. 

745 

746 Notes 

747 ----- 

748 This method is a no-op. 

749 """ 

750 return 

751 

752 def _overrideTransferMode(self, *datasets: Any, transfer: str | None = None) -> str | None: 

753 # Docstring is inherited from base class 

754 return transfer 

755 

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

757 # Docstring is inherited from base class 

758 return 

759 

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

761 # Docstring is inherited from base class 

762 return self.constraints.getLookupKeys() 

763 

764 def needs_expanded_data_ids( 

765 self, 

766 transfer: str | None, 

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

768 ) -> bool: 

769 # Docstring inherited. 

770 return False 

771 

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

773 # Docstring inherited from the base class. 

774 return 

775 

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

777 # Docstring inherited from the base class. 

778 

779 # In-memory Datastore records cannot be exported or imported 

780 return {} 

781 

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

783 # Docstring inherited from the base class. 

784 

785 # In-memory Datastore records cannot be exported or imported 

786 return {} 

787 

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

789 # Docstring inherited from the base class. 

790 return {}