Coverage for python/lsst/daf/butler/datastore/_datastore.py: 96%

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"""Support for generic data stores.""" 

29 

30from __future__ import annotations 

31 

32from .record_data import DatastoreRecordTable 

33 

34__all__ = ( 

35 "DatasetRefURIs", 

36 "Datastore", 

37 "DatastoreConfig", 

38 "DatastoreOpaqueTable", 

39 "DatastoreTransaction", 

40 "DatastoreValidationError", 

41 "DuplicateIdError", 

42 "NullDatastore", 

43) 

44 

45import contextlib 

46import dataclasses 

47import logging 

48import time 

49from abc import ABCMeta, abstractmethod 

50from collections import abc, defaultdict 

51from collections.abc import Callable, Collection, Iterable, Iterator, Mapping 

52from typing import TYPE_CHECKING, Any, ClassVar 

53 

54from lsst.utils import doImportType 

55 

56from .._config import Config, ConfigSubset 

57from .._exceptions import DatasetTypeNotSupportedError, ValidationError 

58from .._file_dataset import FileDataset 

59from .._storage_class import StorageClassFactory 

60from ._transfer import FileTransferMap, FileTransferSource 

61from .constraints import Constraints 

62 

63if TYPE_CHECKING: 

64 from lsst.resources import ResourcePath, ResourcePathExpression 

65 

66 from .. import ddl 

67 from .._config_support import LookupKey 

68 from .._dataset_provenance import DatasetProvenance 

69 from .._dataset_ref import DatasetId, DatasetRef 

70 from .._dataset_type import DatasetType 

71 from .._storage_class import StorageClass 

72 from ..datastores.file_datastore.get import DatasetLocationInformation 

73 from ..datastores.file_datastore.retrieve_artifacts import ArtifactIndexInfo 

74 from ..registry.interfaces import DatasetIdRef, DatastoreRegistryBridgeManager 

75 from .record_data import DatastoreRecordData 

76 from .stored_file_info import StoredDatastoreItemInfo 

77 

78_LOG = logging.getLogger(__name__) 

79 

80 

81class DatastoreConfig(ConfigSubset): 

82 """Configuration for Datastores.""" 

83 

84 component = "datastore" 

85 requiredKeys = ("cls",) 

86 defaultConfigFile = "datastore.yaml" 

87 

88 

89class DatastoreValidationError(ValidationError): 

90 """There is a problem with the Datastore configuration.""" 

91 

92 pass 

93 

94 

95class DuplicateIdError(Exception): 

96 """Exception raised when trying to store dataset with the existing ID.""" 

97 

98 pass 

99 

100 

101@dataclasses.dataclass(frozen=True) 

102class Event: 

103 """Representation of an event that can be rolled back.""" 

104 

105 __slots__ = {"name", "undoFunc", "args", "kwargs"} 

106 name: str 

107 undoFunc: Callable 

108 args: tuple 

109 kwargs: dict 

110 

111 

112@dataclasses.dataclass(frozen=True) 

113class DatastoreOpaqueTable: 

114 """Definition of the opaque table which stores datastore records. 

115 

116 Table definition contains `.ddl.TableSpec` for a table and a class 

117 of a record which must be a subclass of `StoredDatastoreItemInfo`. 

118 """ 

119 

120 __slots__ = {"table_spec", "record_class"} 

121 table_spec: ddl.TableSpec 

122 record_class: type[StoredDatastoreItemInfo] 

123 

124 

125class IngestPrepData: 

126 """A helper base class for `Datastore` ingest implementations. 

127 

128 Datastore implementations will generally need a custom implementation of 

129 this class. 

130 

131 Should be accessed as ``Datastore.IngestPrepData`` instead of via direct 

132 import. 

133 

134 Parameters 

135 ---------- 

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

137 References for the datasets that can be ingested by this datastore. 

138 """ 

139 

140 def __init__(self, refs: Iterable[DatasetRef]): 

141 self.refs = {ref.id: ref for ref in refs} 

142 

143 

144class DatastoreTransaction: 

145 """Keeps a log of `Datastore` activity and allow rollback. 

146 

147 Parameters 

148 ---------- 

149 parent : `DatastoreTransaction`, optional 

150 The parent transaction (if any). 

151 

152 Notes 

153 ----- 

154 This transaction object must be thread safe. 

155 """ 

156 

157 Event: ClassVar[type] = Event 

158 

159 parent: DatastoreTransaction | None 

160 """The parent transaction. (`DatastoreTransaction`, optional)""" 

161 

162 def __init__(self, parent: DatastoreTransaction | None = None): 

163 self.parent = parent 

164 self._log: list[Event] = [] 

165 

166 def registerUndo(self, name: str, undoFunc: Callable, *args: Any, **kwargs: Any) -> None: 

167 """Register event with undo function. 

168 

169 Parameters 

170 ---------- 

171 name : `str` 

172 Name of the event. 

173 undoFunc : `~collections.abc.Callable` 

174 Function to undo this event. 

175 *args : `tuple` 

176 Positional arguments to ``undoFunc``. 

177 **kwargs 

178 Keyword arguments to ``undoFunc``. 

179 """ 

180 self._log.append(self.Event(name, undoFunc, args, kwargs)) 

181 

182 @contextlib.contextmanager 

183 def undoWith(self, name: str, undoFunc: Callable, *args: Any, **kwargs: Any) -> Iterator[None]: 

184 """Register undo function if nested operation succeeds. 

185 

186 Calls `registerUndo`. 

187 

188 This can be used to wrap individual undo-able statements within a 

189 DatastoreTransaction block. Multiple statements that can fail 

190 separately should not be part of the same `undoWith` block. 

191 

192 All arguments are forwarded directly to `registerUndo`. 

193 

194 Parameters 

195 ---------- 

196 name : `str` 

197 The name to associate with this event. 

198 undoFunc : `~collections.abc.Callable` 

199 Function to undo this event. 

200 *args : `tuple` 

201 Positional arguments for ``undoFunc``. 

202 **kwargs : `typing.Any` 

203 Keyword arguments for ``undoFunc``. 

204 """ 

205 try: 

206 yield None 

207 except BaseException: 

208 raise 

209 else: 

210 self.registerUndo(name, undoFunc, *args, **kwargs) 

211 

212 def rollback(self) -> None: 

213 """Roll back all events in this transaction.""" 

214 log = logging.getLogger(__name__) 

215 while self._log: 

216 ev = self._log.pop() 

217 try: 

218 log.debug( 

219 "Rolling back transaction: %s: %s(%s,%s)", 

220 ev.name, 

221 ev.undoFunc, 

222 ",".join(str(a) for a in ev.args), 

223 ",".join(f"{k}={v}" for k, v in ev.kwargs.items()), 

224 ) 

225 except Exception: 

226 # In case we had a problem in stringification of arguments 

227 log.warning("Rolling back transaction: %s", ev.name) 

228 try: 

229 ev.undoFunc(*ev.args, **ev.kwargs) 

230 except BaseException as e: 

231 # Deliberately swallow error that may occur in unrolling 

232 log.warning("Exception: %s caught while unrolling: %s", e, ev.name) 

233 pass 

234 

235 def commit(self) -> None: 

236 """Commit this transaction.""" 

237 if self.parent is None: 

238 # Just forget about the events, they have already happened. 

239 return 

240 else: 

241 # We may still want to events from this transaction as part of 

242 # the parent. 

243 self.parent._log.extend(self._log) 

244 

245 

246@dataclasses.dataclass 

247class DatasetRefURIs(abc.Sequence): 

248 """Represents the primary and component ResourcePath(s) associated with a 

249 DatasetRef. 

250 

251 This is used in places where its members used to be represented as a tuple 

252 (``primaryURI``, ``componentURIs``). To maintain backward compatibility 

253 this inherits from Sequence and so instances can be treated as a two-item 

254 tuple. 

255 

256 Parameters 

257 ---------- 

258 primaryURI : `lsst.resources.ResourcePath` or `None`, optional 

259 The URI to the primary artifact associated with this dataset. If the 

260 dataset was disassembled within the datastore this may be `None`. 

261 componentURIs : `dict` [`str`, `~lsst.resources.ResourcePath`] or `None` 

262 The URIs to any components associated with the dataset artifact 

263 indexed by component name. This can be empty if there are no 

264 components. 

265 """ 

266 

267 def __init__( 

268 self, 

269 primaryURI: ResourcePath | None = None, 

270 componentURIs: dict[str, ResourcePath] | None = None, 

271 ): 

272 self.primaryURI = primaryURI 

273 self.componentURIs = componentURIs or {} 

274 

275 def __getitem__(self, index: Any) -> Any: 

276 """Get primaryURI and componentURIs by index. 

277 

278 Provides support for tuple-like access. 

279 """ 

280 if index == 0: 

281 return self.primaryURI 

282 elif index == 1: 

283 return self.componentURIs 

284 raise IndexError("list index out of range") 

285 

286 def __len__(self) -> int: 

287 """Get the number of data members. 

288 

289 Provides support for tuple-like access. 

290 """ 

291 return 2 

292 

293 def __repr__(self) -> str: 

294 return f"DatasetRefURIs({repr(self.primaryURI)}, {repr(self.componentURIs)})" 

295 

296 def iter_all(self) -> Iterator[ResourcePath]: 

297 """Iterate over all URIs without regard to whether they are primary 

298 or component. 

299 """ 

300 if self.primaryURI is not None: 

301 yield self.primaryURI 

302 yield from self.componentURIs.values() 

303 

304 

305class Datastore(FileTransferSource, metaclass=ABCMeta): 

306 """Datastore interface. 

307 

308 Parameters 

309 ---------- 

310 config : `DatastoreConfig` or `str` 

311 Load configuration either from an existing config instance or by 

312 referring to a configuration file. 

313 bridgeManager : `DatastoreRegistryBridgeManager` 

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

315 

316 See Also 

317 -------- 

318 lsst.daf.butler.Butler 

319 """ 

320 

321 defaultConfigFile: ClassVar[str | None] = None 

322 """Path to configuration defaults. Accessed within the ``config`` resource 

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

324 """ 

325 

326 containerKey: ClassVar[str | None] = None 

327 """Name of the key containing a list of subconfigurations that also 

328 need to be merged with defaults and will likely use different Python 

329 datastore classes (but all using DatastoreConfig). Assumed to be a 

330 list of configurations that can be represented in a DatastoreConfig 

331 and containing a "cls" definition. None indicates that no containers 

332 are expected in this Datastore.""" 

333 

334 isEphemeral: bool = False 

335 """Indicate whether this Datastore is ephemeral or not. An ephemeral 

336 datastore is one where the contents of the datastore will not exist 

337 across process restarts. This value can change per-instance.""" 

338 

339 config: DatastoreConfig 

340 """Configuration used to create Datastore.""" 

341 

342 name: str 

343 """Label associated with this Datastore.""" 

344 

345 storageClassFactory: StorageClassFactory 

346 """Factory for creating storage class instances from name.""" 

347 

348 constraints: Constraints 

349 """Constraints to apply when putting datasets into the datastore.""" 

350 

351 # MyPy does not like for this to be annotated as any kind of type, because 

352 # it can't do static checking on type variables that can change at runtime. 

353 IngestPrepData: ClassVar[Any] = IngestPrepData 

354 """Helper base class for ingest implementations. 

355 """ 

356 

357 @classmethod 

358 @abstractmethod 

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

360 """Set filesystem-dependent config options for this datastore. 

361 

362 The options will be appropriate for a new empty repository with the 

363 given root. 

364 

365 Parameters 

366 ---------- 

367 root : `str` 

368 Filesystem path to the root of the data repository. 

369 config : `Config` 

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

371 this component will be updated. Will not expand 

372 defaults. 

373 full : `Config` 

374 A complete config with all defaults expanded that can be 

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

376 modified by this method. 

377 Repository-specific options that should not be obtained 

378 from defaults when Butler instances are constructed 

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

380 overwrite : `bool`, optional 

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

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

383 ``root``. 

384 

385 Notes 

386 ----- 

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

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

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

390 """ 

391 raise NotImplementedError() 

392 

393 @staticmethod 

394 def fromConfig( 

395 config: Config, 

396 bridgeManager: DatastoreRegistryBridgeManager, 

397 butlerRoot: ResourcePathExpression | None = None, 

398 ) -> Datastore: 

399 """Create datastore from type specified in config file. 

400 

401 Parameters 

402 ---------- 

403 config : `Config` or `~lsst.resources.ResourcePathExpression` 

404 Configuration instance. 

405 bridgeManager : `DatastoreRegistryBridgeManager` 

406 Object that manages the interface between `Registry` and 

407 datastores. 

408 butlerRoot : `str`, optional 

409 Butler root directory. 

410 """ 

411 config = DatastoreConfig(config) 

412 cls = doImportType(config["cls"]) 

413 if not issubclass(cls, Datastore): 413 ↛ 414line 413 didn't jump to line 414 because the condition on line 413 was never true

414 raise TypeError(f"Imported child class {config['cls']} is not a Datastore") 

415 return cls._create_from_config(config=config, bridgeManager=bridgeManager, butlerRoot=butlerRoot) 

416 

417 def __init__( 

418 self, 

419 config: DatastoreConfig, 

420 bridgeManager: DatastoreRegistryBridgeManager, 

421 ): 

422 self.config = config 

423 self.name = "ABCDataStore" 

424 self._transaction: DatastoreTransaction | None = None 

425 

426 # All Datastores need storage classes and constraints 

427 self.storageClassFactory = StorageClassFactory() 

428 

429 # And read the constraints list 

430 constraintsConfig = self.config.get("constraints") 

431 self.constraints = Constraints(constraintsConfig, universe=bridgeManager.universe) 

432 

433 @classmethod 

434 @abstractmethod 

435 def _create_from_config( 

436 cls, 

437 config: DatastoreConfig, 

438 bridgeManager: DatastoreRegistryBridgeManager, 

439 butlerRoot: ResourcePathExpression | None, 

440 ) -> Datastore: 

441 """`Datastore`.``fromConfig`` calls this to instantiate Datastore 

442 subclasses. This is the primary constructor for the individual 

443 Datastore subclasses. 

444 """ 

445 raise NotImplementedError() 

446 

447 @abstractmethod 

448 def clone(self, bridgeManager: DatastoreRegistryBridgeManager) -> Datastore: 

449 """Make an independent copy of this Datastore with a different 

450 `DatastoreRegistryBridgeManager` instance. 

451 

452 Parameters 

453 ---------- 

454 bridgeManager : `DatastoreRegistryBridgeManager` 

455 New `DatastoreRegistryBridgeManager` object to use when 

456 instantiating managers. 

457 

458 Returns 

459 ------- 

460 datastore : `Datastore` 

461 New `Datastore` instance with the same configuration as the 

462 existing instance. 

463 """ 

464 raise NotImplementedError() 

465 

466 def __str__(self) -> str: 

467 return self.name 

468 

469 def __repr__(self) -> str: 

470 return self.name 

471 

472 @property 

473 def names(self) -> tuple[str, ...]: 

474 """Names associated with this datastore returned as a list. 

475 

476 Can be different to ``name`` for a chaining datastore. 

477 """ 

478 # Default implementation returns solely the name itself 

479 return (self.name,) 

480 

481 @property 

482 def roots(self) -> dict[str, ResourcePath | None]: 

483 """Return the root URIs for each named datastore. 

484 

485 Mapping from datastore name to root URI. The URI can be `None` 

486 if a datastore has no concept of a root URI. 

487 (`dict` [`str`, `lsst.resources.ResourcePath` | `None`]) 

488 """ 

489 return {self.name: None} 

490 

491 @contextlib.contextmanager 

492 def transaction(self) -> Iterator[DatastoreTransaction]: 

493 """Context manager supporting `Datastore` transactions. 

494 

495 Transactions can be nested, and are to be used in combination with 

496 `Registry.transaction`. 

497 """ 

498 self._transaction = DatastoreTransaction(self._transaction) 

499 try: 

500 yield self._transaction 

501 except BaseException: 

502 self._transaction.rollback() 

503 raise 

504 else: 

505 self._transaction.commit() 

506 self._transaction = self._transaction.parent 

507 

508 def _set_trust_mode(self, mode: bool) -> None: 

509 """Set the trust mode for this datastore. 

510 

511 Parameters 

512 ---------- 

513 mode : `bool` 

514 If `True`, get requests will be attempted even if the datastore 

515 does not know about the dataset. 

516 

517 Notes 

518 ----- 

519 This is a private method to indicate that trust mode might be a 

520 transitory property that we do not want to make fully public. For now 

521 only a `~lsst.daf.butler.datastores.FileDatastore` understands this 

522 concept. By default this method does nothing. 

523 """ 

524 return 

525 

526 @abstractmethod 

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

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

529 

530 Does not check for existence of any artifact. 

531 

532 Parameters 

533 ---------- 

534 ref : `DatasetRef` 

535 Reference to the required dataset. 

536 

537 Returns 

538 ------- 

539 exists : `bool` 

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

541 """ 

542 raise NotImplementedError() 

543 

544 def knows_these(self, refs: Iterable[DatasetRef]) -> dict[DatasetRef, bool]: 

545 """Check which of the given datasets are known to this datastore. 

546 

547 This is like ``mexist()`` but does not check that the file exists. 

548 

549 Parameters 

550 ---------- 

551 refs : `~collections.abc.Iterable` [ `DatasetRef` ] 

552 The datasets to check. 

553 

554 Returns 

555 ------- 

556 exists : `dict` [`DatasetRef`, `bool`] 

557 Mapping of dataset to boolean indicating whether the dataset 

558 is known to the datastore. 

559 """ 

560 # Non-optimized default calls knows() repeatedly. 

561 return {ref: self.knows(ref) for ref in refs} 

562 

563 def mexists( 

564 self, refs: Iterable[DatasetRef], artifact_existence: dict[ResourcePath, bool] | None = None 

565 ) -> dict[DatasetRef, bool]: 

566 """Check the existence of multiple datasets at once. 

567 

568 Parameters 

569 ---------- 

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

571 The datasets to be checked. 

572 artifact_existence : `dict` [`lsst.resources.ResourcePath`, `bool`] 

573 Optional mapping of datastore artifact to existence. Updated by 

574 this method with details of all artifacts tested. Can be `None` 

575 if the caller is not interested. 

576 

577 Returns 

578 ------- 

579 existence : `dict` of [`DatasetRef`, `bool`] 

580 Mapping from dataset to boolean indicating existence. 

581 """ 

582 existence: dict[DatasetRef, bool] = {} 

583 # Non-optimized default. 

584 for ref in refs: 

585 existence[ref] = self.exists(ref) 

586 return existence 

587 

588 @abstractmethod 

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

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

591 

592 Parameters 

593 ---------- 

594 datasetRef : `DatasetRef` 

595 Reference to the required dataset. 

596 

597 Returns 

598 ------- 

599 exists : `bool` 

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

601 """ 

602 raise NotImplementedError("Must be implemented by subclass") 

603 

604 @abstractmethod 

605 def get( 

606 self, 

607 datasetRef: DatasetRef, 

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

609 storageClass: StorageClass | str | None = None, 

610 ) -> Any: 

611 """Load an in-memory dataset from the store. 

612 

613 Parameters 

614 ---------- 

615 datasetRef : `DatasetRef` 

616 Reference to the required Dataset. 

617 parameters : `dict` 

618 `StorageClass`-specific parameters that specify a slice of the 

619 Dataset to be loaded. 

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

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

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

623 the dataset type definition for this dataset. Specifying a 

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

625 This type must be compatible with the original type. 

626 

627 Returns 

628 ------- 

629 inMemoryDataset : `object` 

630 Requested Dataset or slice thereof as an in-memory dataset. 

631 """ 

632 raise NotImplementedError("Must be implemented by subclass") 

633 

634 def prepare_get_for_external_client(self, ref: DatasetRef) -> list[DatasetLocationInformation] | None: 

635 """Retrieve data that can be used to execute a ``get()``. 

636 

637 Parameters 

638 ---------- 

639 ref : `DatasetRef` 

640 Reference to the required dataset. 

641 

642 Returns 

643 ------- 

644 payload : `list` [ `DatasetLocationInformation` ] | `None` 

645 Information needed to perform a get() operation. Returns `None` if 

646 the dataset is not known to this datastore. 

647 """ 

648 raise NotImplementedError() 

649 

650 @abstractmethod 

651 def put( 

652 self, inMemoryDataset: Any, datasetRef: DatasetRef, provenance: DatasetProvenance | None = None 

653 ) -> None: 

654 """Write an in-memory dataset with a given `DatasetRef` to the store. 

655 

656 Parameters 

657 ---------- 

658 inMemoryDataset : `object` 

659 The Dataset to store. 

660 datasetRef : `DatasetRef` 

661 Reference to the associated Dataset. 

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

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

664 Not supported by all serialization mechanisms. 

665 

666 Raises 

667 ------ 

668 DuplicateIdError 

669 Raised if datastore already has a dataset with the same dataset ID. 

670 """ 

671 raise NotImplementedError("Must be implemented by subclass") 

672 

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

674 """Allow ingest transfer mode to be defaulted based on datasets. 

675 

676 Parameters 

677 ---------- 

678 *datasets : `FileDataset` 

679 Each positional argument is a struct containing information about 

680 a file to be ingested, including its path (either absolute or 

681 relative to the datastore root, if applicable), a complete 

682 `DatasetRef` (with ``dataset_id not None``), and optionally a 

683 formatter class or its fully-qualified string name. If a formatter 

684 is not provided, this method should populate that attribute with 

685 the formatter the datastore would use for `put`. Subclasses are 

686 also permitted to modify the path attribute (typically to put it 

687 in what the datastore considers its standard form). 

688 transfer : `str`, optional 

689 How (and whether) the dataset should be added to the datastore. 

690 See `ingest` for details of transfer modes. 

691 

692 Returns 

693 ------- 

694 newTransfer : `str` 

695 Transfer mode to use. Will be identical to the supplied transfer 

696 mode unless "auto" is used. 

697 """ 

698 if transfer != "auto": 

699 return transfer 

700 raise RuntimeError(f"{transfer} is not allowed without specialization.") 

701 

702 def _prepIngest(self, *datasets: FileDataset, transfer: str | None = None) -> IngestPrepData: 

703 """Process datasets to identify which ones can be ingested. 

704 

705 Parameters 

706 ---------- 

707 *datasets : `FileDataset` 

708 Each positional argument is a struct containing information about 

709 a file to be ingested, including its path (either absolute or 

710 relative to the datastore root, if applicable), a complete 

711 `DatasetRef` (with ``dataset_id not None``), and optionally a 

712 formatter class or its fully-qualified string name. If a formatter 

713 is not provided, this method should populate that attribute with 

714 the formatter the datastore would use for `put`. Subclasses are 

715 also permitted to modify the path attribute (typically to put it 

716 in what the datastore considers its standard form). 

717 transfer : `str`, optional 

718 How (and whether) the dataset should be added to the datastore. 

719 See `ingest` for details of transfer modes. 

720 

721 Returns 

722 ------- 

723 data : `IngestPrepData` 

724 An instance of a subclass of `IngestPrepData`, used to pass 

725 arbitrary data from `_prepIngest` to `_finishIngest`. This should 

726 include only the datasets this datastore can actually ingest; 

727 others should be silently ignored (`Datastore.ingest` will inspect 

728 `IngestPrepData.refs` and raise `DatasetTypeNotSupportedError` if 

729 necessary). 

730 

731 Raises 

732 ------ 

733 NotImplementedError 

734 Raised if the datastore does not support the given transfer mode 

735 (including the case where ingest is not supported at all). 

736 FileNotFoundError 

737 Raised if one of the given files does not exist. 

738 FileExistsError 

739 Raised if transfer is not `None` but the (internal) location the 

740 file would be moved to is already occupied. 

741 DuplicateIdError 

742 Raised if dataset ID of any dataset already exists in this 

743 datastore. 

744 

745 Notes 

746 ----- 

747 This method (along with `_finishIngest`) should be implemented by 

748 subclasses to provide ingest support instead of implementing `ingest` 

749 directly. 

750 

751 `_prepIngest` should not modify the data repository or given files in 

752 any way; all changes should be deferred to `_finishIngest`. 

753 

754 When possible, exceptions should be raised in `_prepIngest` instead of 

755 `_finishIngest`. `NotImplementedError` exceptions that indicate that 

756 the transfer mode is not supported must be raised by `_prepIngest` 

757 instead of `_finishIngest`. 

758 """ 

759 raise NotImplementedError(f"Datastore {self} does not support direct file-based ingest.") 

760 

761 def _finishIngest( 

762 self, prepData: IngestPrepData, *, transfer: str | None = None, record_validation_info: bool = True 

763 ) -> None: 

764 """Complete an ingest operation. 

765 

766 Parameters 

767 ---------- 

768 prepData : `IngestPrepData` 

769 An instance of a subclass of `IngestPrepData`. Guaranteed to be 

770 the direct result of a call to `_prepIngest` on this datastore. 

771 transfer : `str`, optional 

772 How (and whether) the dataset should be added to the datastore. 

773 See `ingest` for details of transfer modes. 

774 record_validation_info : `bool`, optional 

775 If `True`, the default, the datastore can record validation 

776 information associated with the file. If `False` the datastore 

777 will not attempt to track any information such as checksums 

778 or file sizes. This can be useful if such information is tracked 

779 in an external system or if the file is to be compressed in place. 

780 It is up to the datastore whether this parameter is relevant. 

781 

782 Raises 

783 ------ 

784 FileNotFoundError 

785 Raised if one of the given files does not exist. 

786 FileExistsError 

787 Raised if transfer is not `None` but the (internal) location the 

788 file would be moved to is already occupied. 

789 

790 Notes 

791 ----- 

792 This method (along with `_prepIngest`) should be implemented by 

793 subclasses to provide ingest support instead of implementing `ingest` 

794 directly. 

795 """ 

796 raise NotImplementedError(f"Datastore {self} does not support direct file-based ingest.") 

797 

798 def ingest( 

799 self, *datasets: FileDataset, transfer: str | None = None, record_validation_info: bool = True 

800 ) -> None: 

801 """Ingest one or more files into the datastore. 

802 

803 Parameters 

804 ---------- 

805 *datasets : `FileDataset` 

806 Each positional argument is a struct containing information about 

807 a file to be ingested, including its path (either absolute or 

808 relative to the datastore root, if applicable), a complete 

809 `DatasetRef` (with ``dataset_id not None``), and optionally a 

810 formatter class or its fully-qualified string name. If a formatter 

811 is not provided, the one the datastore would use for ``put`` on 

812 that dataset is assumed. 

813 transfer : `str`, optional 

814 How (and whether) the dataset should be added to the datastore. 

815 If `None` (default), the file must already be in a location 

816 appropriate for the datastore (e.g. within its root directory), 

817 and will not be modified. Other choices include "move", "copy", 

818 "link", "symlink", "relsymlink", and "hardlink". "link" is a 

819 special transfer mode that will first try to make a hardlink and 

820 if that fails a symlink will be used instead. "relsymlink" creates 

821 a relative symlink rather than use an absolute path. 

822 Most datastores do not support all transfer modes. 

823 "auto" is a special option that will let the 

824 data store choose the most natural option for itself. 

825 record_validation_info : `bool`, optional 

826 If `True`, the default, the datastore can record validation 

827 information associated with the file. If `False` the datastore 

828 will not attempt to track any information such as checksums 

829 or file sizes. This can be useful if such information is tracked 

830 in an external system or if the file is to be compressed in place. 

831 It is up to the datastore whether this parameter is relevant. 

832 

833 Returns 

834 ------- 

835 None 

836 

837 Raises 

838 ------ 

839 NotImplementedError 

840 Raised if the datastore does not support the given transfer mode 

841 (including the case where ingest is not supported at all). 

842 DatasetTypeNotSupportedError 

843 Raised if one or more files to be ingested have a dataset type that 

844 is not supported by the datastore. 

845 FileNotFoundError 

846 Raised if one of the given files does not exist. 

847 FileExistsError 

848 Raised if transfer is not `None` but the (internal) location the 

849 file would be moved to is already occupied. 

850 DuplicateIdError 

851 Raised if dataset ID of any dataset already exists in this 

852 datastore. 

853 

854 Notes 

855 ----- 

856 Subclasses should implement `_prepIngest` and `_finishIngest` instead 

857 of implementing `ingest` directly. Datastores that hold and 

858 delegate to child datastores may want to call those methods as well. 

859 

860 Subclasses are encouraged to document their supported transfer modes 

861 in their class documentation. 

862 """ 

863 # Allow a datastore to select a default transfer mode 

864 transfer = self._overrideTransferMode(*datasets, transfer=transfer) 

865 prepData = self._prepIngest(*datasets, transfer=transfer) 

866 refs = {ref.id: ref for dataset in datasets for ref in dataset.refs} 

867 if refs.keys() != prepData.refs.keys(): 

868 unsupported = refs.keys() - prepData.refs.keys() 

869 # Group unsupported refs by DatasetType for an informative 

870 # but still concise error message. 

871 byDatasetType = defaultdict(list) 

872 for datasetId in unsupported: 

873 ref = refs[datasetId] 

874 byDatasetType[ref.datasetType].append(ref) 

875 raise DatasetTypeNotSupportedError( 

876 "DatasetType(s) not supported in ingest: " 

877 + ", ".join(f"{k.name} ({len(v)} dataset(s))" for k, v in byDatasetType.items()) 

878 ) 

879 self._finishIngest(prepData, transfer=transfer, record_validation_info=record_validation_info) 

880 

881 def transfer_from( 

882 self, 

883 source_records: FileTransferMap, 

884 refs: Collection[DatasetRef], 

885 transfer: str = "auto", 

886 artifact_existence: dict[ResourcePath, bool] | None = None, 

887 dry_run: bool = False, 

888 ) -> tuple[set[DatasetRef], set[DatasetRef]]: 

889 """Transfer dataset artifacts from another datastore to this one. 

890 

891 Parameters 

892 ---------- 

893 source_records : `FileTransferMap` 

894 The artifacts to be transferred into this datastore. 

895 refs : `~collections.abc.Collection` of `DatasetRef` 

896 The datasets to transfer from the source datastore. 

897 transfer : `str`, optional 

898 How (and whether) the dataset should be added to the datastore. 

899 Choices include "move", "copy", 

900 "link", "symlink", "relsymlink", and "hardlink". "link" is a 

901 special transfer mode that will first try to make a hardlink and 

902 if that fails a symlink will be used instead. "relsymlink" creates 

903 a relative symlink rather than use an absolute path. 

904 Most datastores do not support all transfer modes. 

905 "auto" (the default) is a special option that will let the 

906 data store choose the most natural option for itself. 

907 If the source location and transfer location are identical the 

908 transfer mode will be ignored. 

909 artifact_existence : `dict` [`lsst.resources.ResourcePath`, `bool`] 

910 Optional mapping of datastore artifact to existence. Updated by 

911 this method with details of all artifacts tested. Can be `None` 

912 if the caller is not interested. 

913 dry_run : `bool`, optional 

914 Process the supplied source refs without updating the target 

915 datastore. 

916 

917 Returns 

918 ------- 

919 accepted : `set` [`DatasetRef`] 

920 The datasets that were transferred. 

921 rejected : `set` [`DatasetRef`] 

922 The datasets that were rejected due to a constraints violation. 

923 

924 Raises 

925 ------ 

926 TypeError 

927 Raised if the two datastores are not compatible. 

928 """ 

929 raise NotImplementedError(f"Datastore {type(self)} does not implement a transfer_from method.") 

930 

931 def getManyURIs( 

932 self, 

933 refs: Iterable[DatasetRef], 

934 predict: bool = False, 

935 allow_missing: bool = False, 

936 ) -> dict[DatasetRef, DatasetRefURIs]: 

937 """Return URIs associated with many datasets. 

938 

939 Parameters 

940 ---------- 

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

942 References to the required datasets. 

943 predict : `bool`, optional 

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

945 been written. 

946 allow_missing : `bool` 

947 If `False`, and ``predict`` is `False`, will raise if a 

948 `DatasetRef` does not exist. 

949 

950 Returns 

951 ------- 

952 URIs : `dict` of [`DatasetRef`, `DatasetRefUris`] 

953 A dict of primary and component URIs, indexed by the passed-in 

954 refs. 

955 

956 Raises 

957 ------ 

958 FileNotFoundError 

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

960 guessing is not allowed. 

961 

962 Notes 

963 ----- 

964 In file-based datastores, getManyURIs does not check that the file is 

965 really there, it's assuming it is if datastore is aware of the file 

966 then it actually exists. 

967 """ 

968 uris: dict[DatasetRef, DatasetRefURIs] = {} 

969 missing_refs = [] 

970 for ref in refs: 

971 try: 

972 uris[ref] = self.getURIs(ref, predict=predict) 

973 except FileNotFoundError: 

974 missing_refs.append(ref) 

975 if missing_refs and not allow_missing: 

976 num_missing = len(missing_refs) 

977 raise FileNotFoundError( 

978 f"Missing {num_missing} refs from datastore out of " 

979 f"{num_missing + len(uris)} and predict=False." 

980 ) 

981 return uris 

982 

983 @abstractmethod 

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

985 """Return URIs associated with dataset. 

986 

987 Parameters 

988 ---------- 

989 datasetRef : `DatasetRef` 

990 Reference to the required dataset. 

991 predict : `bool`, optional 

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

993 it should return a predicted URI or not. 

994 

995 Returns 

996 ------- 

997 uris : `DatasetRefURIs` 

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

999 the dataset was disassembled within the datastore this may be 

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

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

1002 """ 

1003 raise NotImplementedError() 

1004 

1005 @abstractmethod 

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

1007 """URI to the Dataset. 

1008 

1009 Parameters 

1010 ---------- 

1011 datasetRef : `DatasetRef` 

1012 Reference to the required Dataset. 

1013 predict : `bool` 

1014 If `True` attempt to predict the URI for a dataset if it does 

1015 not exist in datastore. 

1016 

1017 Returns 

1018 ------- 

1019 uri : `str` 

1020 URI string pointing to the Dataset within the datastore. If the 

1021 Dataset does not exist in the datastore, the URI may be a guess. 

1022 If the datastore does not have entities that relate well 

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

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

1025 

1026 Raises 

1027 ------ 

1028 FileNotFoundError 

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

1030 guessing is not allowed. 

1031 """ 

1032 raise NotImplementedError("Must be implemented by subclass") 

1033 

1034 @abstractmethod 

1035 def retrieveArtifacts( 

1036 self, 

1037 refs: Iterable[DatasetRef], 

1038 destination: ResourcePath, 

1039 transfer: str = "auto", 

1040 preserve_path: bool = True, 

1041 overwrite: bool = False, 

1042 write_index: bool = True, 

1043 add_prefix: bool = False, 

1044 ) -> dict[ResourcePath, ArtifactIndexInfo]: 

1045 """Retrieve the artifacts associated with the supplied refs. 

1046 

1047 Parameters 

1048 ---------- 

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

1050 The datasets for which artifacts are to be retrieved. 

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

1052 be resolved. 

1053 destination : `lsst.resources.ResourcePath` 

1054 Location to write the artifacts. 

1055 transfer : `str`, optional 

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

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

1058 "move" is not allowed. 

1059 preserve_path : `bool`, optional 

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

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

1062 is used. 

1063 overwrite : `bool`, optional 

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

1065 destination. 

1066 write_index : `bool`, optional 

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