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

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 

32__all__ = ( 

33 "DatasetRefURIs", 

34 "Datastore", 

35 "DatastoreConfig", 

36 "DatastoreOpaqueTable", 

37 "DatastoreTransaction", 

38 "DatastoreValidationError", 

39 "NullDatastore", 

40) 

41 

42import contextlib 

43import dataclasses 

44import logging 

45import time 

46from abc import ABCMeta, abstractmethod 

47from collections import abc, defaultdict 

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

49from typing import TYPE_CHECKING, Any, ClassVar 

50 

51from lsst.utils import doImportType 

52 

53from .._config import Config, ConfigSubset 

54from .._exceptions import DatasetTypeNotSupportedError, ValidationError 

55from .._file_dataset import FileDataset 

56from .._storage_class import StorageClassFactory 

57from ._transfer import FileTransferMap, FileTransferSource 

58from .constraints import Constraints 

59 

60if TYPE_CHECKING: 

61 from lsst.resources import ResourcePath, ResourcePathExpression 

62 

63 from .. import ddl 

64 from .._config_support import LookupKey 

65 from .._dataset_provenance import DatasetProvenance 

66 from .._dataset_ref import DatasetId, DatasetRef 

67 from .._dataset_type import DatasetType 

68 from .._storage_class import StorageClass 

69 from ..datastores.file_datastore.get import DatasetLocationInformation 

70 from ..datastores.file_datastore.retrieve_artifacts import ArtifactIndexInfo 

71 from ..registry.interfaces import DatasetIdRef, DatastoreRegistryBridgeManager 

72 from .record_data import DatastoreRecordData 

73 from .stored_file_info import StoredDatastoreItemInfo 

74 

75_LOG = logging.getLogger(__name__) 

76 

77 

78class DatastoreConfig(ConfigSubset): 

79 """Configuration for Datastores.""" 

80 

81 component = "datastore" 

82 requiredKeys = ("cls",) 

83 defaultConfigFile = "datastore.yaml" 

84 

85 

86class DatastoreValidationError(ValidationError): 

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

88 

89 pass 

90 

91 

92@dataclasses.dataclass(frozen=True) 

93class Event: 

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

95 

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

97 name: str 

98 undoFunc: Callable 

99 args: tuple 

100 kwargs: dict 

101 

102 

103@dataclasses.dataclass(frozen=True) 

104class DatastoreOpaqueTable: 

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

106 

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

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

109 """ 

110 

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

112 table_spec: ddl.TableSpec 

113 record_class: type[StoredDatastoreItemInfo] 

114 

115 

116class IngestPrepData: 

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

118 

119 Datastore implementations will generally need a custom implementation of 

120 this class. 

121 

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

123 import. 

124 

125 Parameters 

126 ---------- 

127 refs : iterable of `DatasetRef` 

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

129 """ 

130 

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

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

133 

134 

135class DatastoreTransaction: 

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

137 

138 Parameters 

139 ---------- 

140 parent : `DatastoreTransaction`, optional 

141 The parent transaction (if any). 

142 

143 Notes 

144 ----- 

145 This transaction object must be thread safe. 

146 """ 

147 

148 Event: ClassVar[type] = Event 

149 

150 parent: DatastoreTransaction | None 

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

152 

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

154 self.parent = parent 

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

156 

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

158 """Register event with undo function. 

159 

160 Parameters 

161 ---------- 

162 name : `str` 

163 Name of the event. 

164 undoFunc : `~collections.abc.Callable` 

165 Function to undo this event. 

166 *args : `tuple` 

167 Positional arguments to ``undoFunc``. 

168 **kwargs 

169 Keyword arguments to ``undoFunc``. 

170 """ 

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

172 

173 @contextlib.contextmanager 

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

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

176 

177 Calls `registerUndo`. 

178 

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

180 DatastoreTransaction block. Multiple statements that can fail 

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

182 

183 All arguments are forwarded directly to `registerUndo`. 

184 

185 Parameters 

186 ---------- 

187 name : `str` 

188 The name to associate with this event. 

189 undoFunc : `~collections.abc.Callable` 

190 Function to undo this event. 

191 *args : `tuple` 

192 Positional arguments for ``undoFunc``. 

193 **kwargs : `typing.Any` 

194 Keyword arguments for ``undoFunc``. 

195 """ 

196 try: 

197 yield None 

198 except BaseException: 

199 raise 

200 else: 

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

202 

203 def rollback(self) -> None: 

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

205 log = logging.getLogger(__name__) 

206 while self._log: 

207 ev = self._log.pop() 

208 try: 

209 log.debug( 

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

211 ev.name, 

212 ev.undoFunc, 

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

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

215 ) 

216 except Exception: 

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

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

219 try: 

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

221 except BaseException as e: 

222 # Deliberately swallow error that may occur in unrolling 

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

224 pass 

225 

226 def commit(self) -> None: 

227 """Commit this transaction.""" 

228 if self.parent is None: 

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

230 return 

231 else: 

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

233 # the parent. 

234 self.parent._log.extend(self._log) 

235 

236 

237@dataclasses.dataclass 

238class DatasetRefURIs(abc.Sequence): 

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

240 DatasetRef. 

241 

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

243 `(primaryURI, componentURIs)`. To maintain backward compatibility this 

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

245 tuple. 

246 

247 Parameters 

248 ---------- 

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

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

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

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

253 The URIs to any components associated with the dataset artifact 

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

255 components. 

256 """ 

257 

258 def __init__( 

259 self, 

260 primaryURI: ResourcePath | None = None, 

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

262 ): 

263 self.primaryURI = primaryURI 

264 self.componentURIs = componentURIs or {} 

265 

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

267 """Get primaryURI and componentURIs by index. 

268 

269 Provides support for tuple-like access. 

270 """ 

271 if index == 0: 

272 return self.primaryURI 

273 elif index == 1: 

274 return self.componentURIs 

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

276 

277 def __len__(self) -> int: 

278 """Get the number of data members. 

279 

280 Provides support for tuple-like access. 

281 """ 

282 return 2 

283 

284 def __repr__(self) -> str: 

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

286 

287 

288class Datastore(FileTransferSource, metaclass=ABCMeta): 

289 """Datastore interface. 

290 

291 Parameters 

292 ---------- 

293 config : `DatastoreConfig` or `str` 

294 Load configuration either from an existing config instance or by 

295 referring to a configuration file. 

296 bridgeManager : `DatastoreRegistryBridgeManager` 

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

298 

299 See Also 

300 -------- 

301 lsst.daf.butler.Butler 

302 """ 

303 

304 defaultConfigFile: ClassVar[str | None] = None 

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

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

307 """ 

308 

309 containerKey: ClassVar[str | None] = None 

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

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

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

313 list of configurations that can be represented in a DatastoreConfig 

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

315 are expected in this Datastore.""" 

316 

317 isEphemeral: bool = False 

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

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

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

321 

322 config: DatastoreConfig 

323 """Configuration used to create Datastore.""" 

324 

325 name: str 

326 """Label associated with this Datastore.""" 

327 

328 storageClassFactory: StorageClassFactory 

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

330 

331 constraints: Constraints 

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

333 

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

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

336 IngestPrepData: ClassVar[Any] = IngestPrepData 

337 """Helper base class for ingest implementations. 

338 """ 

339 

340 @classmethod 

341 @abstractmethod 

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

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

344 

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

346 given root. 

347 

348 Parameters 

349 ---------- 

350 root : `str` 

351 Filesystem path to the root of the data repository. 

352 config : `Config` 

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

354 this component will be updated. Will not expand 

355 defaults. 

356 full : `Config` 

357 A complete config with all defaults expanded that can be 

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

359 modified by this method. 

360 Repository-specific options that should not be obtained 

361 from defaults when Butler instances are constructed 

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

363 overwrite : `bool`, optional 

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

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

366 ``root``. 

367 

368 Notes 

369 ----- 

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

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

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

373 """ 

374 raise NotImplementedError() 

375 

376 @staticmethod 

377 def fromConfig( 

378 config: Config, 

379 bridgeManager: DatastoreRegistryBridgeManager, 

380 butlerRoot: ResourcePathExpression | None = None, 

381 ) -> Datastore: 

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

383 

384 Parameters 

385 ---------- 

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

387 Configuration instance. 

388 bridgeManager : `DatastoreRegistryBridgeManager` 

389 Object that manages the interface between `Registry` and 

390 datastores. 

391 butlerRoot : `str`, optional 

392 Butler root directory. 

393 """ 

394 config = DatastoreConfig(config) 

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

396 if not issubclass(cls, Datastore): 

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

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

399 

400 def __init__( 

401 self, 

402 config: DatastoreConfig, 

403 bridgeManager: DatastoreRegistryBridgeManager, 

404 ): 

405 self.config = config 

406 self.name = "ABCDataStore" 

407 self._transaction: DatastoreTransaction | None = None 

408 

409 # All Datastores need storage classes and constraints 

410 self.storageClassFactory = StorageClassFactory() 

411 

412 # And read the constraints list 

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

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

415 

416 @classmethod 

417 @abstractmethod 

418 def _create_from_config( 

419 cls, 

420 config: DatastoreConfig, 

421 bridgeManager: DatastoreRegistryBridgeManager, 

422 butlerRoot: ResourcePathExpression | None, 

423 ) -> Datastore: 

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

425 subclasses. This is the primary constructor for the individual 

426 Datastore subclasses. 

427 """ 

428 raise NotImplementedError() 

429 

430 @abstractmethod 

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

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

433 `DatastoreRegistryBridgeManager` instance. 

434 

435 Parameters 

436 ---------- 

437 bridgeManager : `DatastoreRegistryBridgeManager` 

438 New `DatastoreRegistryBridgeManager` object to use when 

439 instantiating managers. 

440 

441 Returns 

442 ------- 

443 datastore : `Datastore` 

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

445 existing instance. 

446 """ 

447 raise NotImplementedError() 

448 

449 def __str__(self) -> str: 

450 return self.name 

451 

452 def __repr__(self) -> str: 

453 return self.name 

454 

455 @property 

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

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

458 

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

460 """ 

461 # Default implementation returns solely the name itself 

462 return (self.name,) 

463 

464 @property 

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

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

467 

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

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

470 (`dict` [`str`, `ResourcePath` | `None`]) 

471 """ 

472 return {self.name: None} 

473 

474 @contextlib.contextmanager 

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

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

477 

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

479 `Registry.transaction`. 

480 """ 

481 self._transaction = DatastoreTransaction(self._transaction) 

482 try: 

483 yield self._transaction 

484 except BaseException: 

485 self._transaction.rollback() 

486 raise 

487 else: 

488 self._transaction.commit() 

489 self._transaction = self._transaction.parent 

490 

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

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

493 

494 Parameters 

495 ---------- 

496 mode : `bool` 

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

498 does not know about the dataset. 

499 

500 Notes 

501 ----- 

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

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

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

505 concept. By default this method does nothing. 

506 """ 

507 return 

508 

509 @abstractmethod 

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

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

512 

513 Does not check for existence of any artifact. 

514 

515 Parameters 

516 ---------- 

517 ref : `DatasetRef` 

518 Reference to the required dataset. 

519 

520 Returns 

521 ------- 

522 exists : `bool` 

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

524 """ 

525 raise NotImplementedError() 

526 

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

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

529 

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

531 

532 Parameters 

533 ---------- 

534 refs : iterable `DatasetRef` 

535 The datasets to check. 

536 

537 Returns 

538 ------- 

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

540 Mapping of dataset to boolean indicating whether the dataset 

541 is known to the datastore. 

542 """ 

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

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

545 

546 def mexists( 

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

548 ) -> dict[DatasetRef, bool]: 

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

550 

551 Parameters 

552 ---------- 

553 refs : iterable of `DatasetRef` 

554 The datasets to be checked. 

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

556 Optional mapping of datastore artifact to existence. Updated by 

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

558 if the caller is not interested. 

559 

560 Returns 

561 ------- 

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

563 Mapping from dataset to boolean indicating existence. 

564 """ 

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

566 # Non-optimized default. 

567 for ref in refs: 

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

569 return existence 

570 

571 @abstractmethod 

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

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

574 

575 Parameters 

576 ---------- 

577 datasetRef : `DatasetRef` 

578 Reference to the required dataset. 

579 

580 Returns 

581 ------- 

582 exists : `bool` 

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

584 """ 

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

586 

587 @abstractmethod 

588 def get( 

589 self, 

590 datasetRef: DatasetRef, 

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

592 storageClass: StorageClass | str | None = None, 

593 ) -> Any: 

594 """Load an `InMemoryDataset` from the store. 

595 

596 Parameters 

597 ---------- 

598 datasetRef : `DatasetRef` 

599 Reference to the required Dataset. 

600 parameters : `dict` 

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

602 Dataset to be loaded. 

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

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

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

606 the dataset type definition for this dataset. Specifying a 

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

608 This type must be compatible with the original type. 

609 

610 Returns 

611 ------- 

612 inMemoryDataset : `object` 

613 Requested Dataset or slice thereof as an InMemoryDataset. 

614 """ 

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

616 

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

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

619 

620 Parameters 

621 ---------- 

622 ref : `DatasetRef` 

623 Reference to the required dataset. 

624 

625 Returns 

626 ------- 

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

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

629 the dataset is not known to this datastore. 

630 """ 

631 raise NotImplementedError() 

632 

633 @abstractmethod 

634 def put( 

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

636 ) -> None: 

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

638 

639 Parameters 

640 ---------- 

641 inMemoryDataset : `object` 

642 The Dataset to store. 

643 datasetRef : `DatasetRef` 

644 Reference to the associated Dataset. 

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

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

647 Not supported by all serialization mechanisms. 

648 """ 

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

650 

651 @abstractmethod 

652 def put_new(self, in_memory_dataset: Any, ref: DatasetRef) -> Mapping[str, DatasetRef]: 

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

654 

655 Parameters 

656 ---------- 

657 in_memory_dataset : `object` 

658 The Dataset to store. 

659 ref : `DatasetRef` 

660 Reference to the associated Dataset. 

661 

662 Returns 

663 ------- 

664 datastore_refs : `~collections.abc.Mapping` [`str`, `DatasetRef`] 

665 Mapping of a datastore name to dataset reference stored in that 

666 datastore, reference will include datastore records. Only 

667 non-ephemeral datastores will appear in this mapping. 

668 """ 

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

670 

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

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

673 

674 Parameters 

675 ---------- 

676 *datasets : `FileDataset` 

677 Each positional argument is a struct containing information about 

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

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

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

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

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

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

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

685 in what the datastore considers its standard form). 

686 transfer : `str`, optional 

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

688 See `ingest` for details of transfer modes. 

689 

690 Returns 

691 ------- 

692 newTransfer : `str` 

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

694 mode unless "auto" is used. 

695 """ 

696 if transfer != "auto": 

697 return transfer 

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

699 

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

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

702 

703 Parameters 

704 ---------- 

705 *datasets : `FileDataset` 

706 Each positional argument is a struct containing information about 

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

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

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

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

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

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

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

714 in what the datastore considers its standard form). 

715 transfer : `str`, optional 

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

717 See `ingest` for details of transfer modes. 

718 

719 Returns 

720 ------- 

721 data : `IngestPrepData` 

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

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

724 include only the datasets this datastore can actually ingest; 

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

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

727 necessary). 

728 

729 Raises 

730 ------ 

731 NotImplementedError 

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

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

734 FileNotFoundError 

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

736 FileExistsError 

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

738 file would be moved to is already occupied. 

739 

740 Notes 

741 ----- 

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

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

744 directly. 

745 

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

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

748 

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

750 `_finishIngest`. `NotImplementedError` exceptions that indicate that 

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

752 instead of `_finishIngest`. 

753 """ 

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

755 

756 def _finishIngest( 

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

758 ) -> None: 

759 """Complete an ingest operation. 

760 

761 Parameters 

762 ---------- 

763 prepData : `IngestPrepData` 

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

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

766 transfer : `str`, optional 

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

768 See `ingest` for details of transfer modes. 

769 record_validation_info : `bool`, optional 

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

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

772 will not attempt to track any information such as checksums 

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

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

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

776 

777 Raises 

778 ------ 

779 FileNotFoundError 

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

781 FileExistsError 

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

783 file would be moved to is already occupied. 

784 

785 Notes 

786 ----- 

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

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

789 directly. 

790 """ 

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

792 

793 def ingest( 

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

795 ) -> None: 

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

797 

798 Parameters 

799 ---------- 

800 *datasets : `FileDataset` 

801 Each positional argument is a struct containing information about 

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

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

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

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

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

807 that dataset is assumed. 

808 transfer : `str`, optional 

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

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

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

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

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

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

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

816 a relative symlink rather than use an absolute path. 

817 Most datastores do not support all transfer modes. 

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

819 data store choose the most natural option for itself. 

820 record_validation_info : `bool`, optional 

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

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

823 will not attempt to track any information such as checksums 

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

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

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

827 

828 Raises 

829 ------ 

830 NotImplementedError 

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

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

833 DatasetTypeNotSupportedError 

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

835 is not supported by the datastore. 

836 FileNotFoundError 

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

838 FileExistsError 

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

840 file would be moved to is already occupied. 

841 

842 Notes 

843 ----- 

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

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

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

847 

848 Subclasses are encouraged to document their supported transfer modes 

849 in their class documentation. 

850 """ 

851 # Allow a datastore to select a default transfer mode 

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

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

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

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

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

857 # Group unsupported refs by DatasetType for an informative 

858 # but still concise error message. 

859 byDatasetType = defaultdict(list) 

860 for datasetId in unsupported: 

861 ref = refs[datasetId] 

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

863 raise DatasetTypeNotSupportedError( 

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

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

866 ) 

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

868 

869 def transfer_from( 

870 self, 

871 source_records: FileTransferMap, 

872 refs: Collection[DatasetRef], 

873 transfer: str = "auto", 

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

875 dry_run: bool = False, 

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

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

878 

879 Parameters 

880 ---------- 

881 source_records : `FileTransferMap` 

882 The artifacts to be transferred into this datastore. 

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

884 The datasets to transfer from the source datastore. 

885 transfer : `str`, optional 

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

887 Choices include "move", "copy", 

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

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

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

891 a relative symlink rather than use an absolute path. 

892 Most datastores do not support all transfer modes. 

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

894 data store choose the most natural option for itself. 

895 If the source location and transfer location are identical the 

896 transfer mode will be ignored. 

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

898 Optional mapping of datastore artifact to existence. Updated by 

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

900 if the caller is not interested. 

901 dry_run : `bool`, optional 

902 Process the supplied source refs without updating the target 

903 datastore. 

904 

905 Returns 

906 ------- 

907 accepted : `set` [`DatasetRef`] 

908 The datasets that were transferred. 

909 rejected : `set` [`DatasetRef`] 

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

911 

912 Raises 

913 ------ 

914 TypeError 

915 Raised if the two datastores are not compatible. 

916 """ 

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

918 

919 def getManyURIs( 

920 self, 

921 refs: Iterable[DatasetRef], 

922 predict: bool = False, 

923 allow_missing: bool = False, 

924 ) -> dict[DatasetRef, DatasetRefURIs]: 

925 """Return URIs associated with many datasets. 

926 

927 Parameters 

928 ---------- 

929 refs : iterable of `DatasetIdRef` 

930 References to the required datasets. 

931 predict : `bool`, optional 

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

933 been written. 

934 allow_missing : `bool` 

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

936 `DatasetRef` does not exist. 

937 

938 Returns 

939 ------- 

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

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

942 refs. 

943 

944 Raises 

945 ------ 

946 FileNotFoundError 

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

948 guessing is not allowed. 

949 

950 Notes 

951 ----- 

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

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

954 then it actually exists. 

955 """ 

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

957 missing_refs = [] 

958 for ref in refs: 

959 try: 

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

961 except FileNotFoundError: 

962 missing_refs.append(ref) 

963 if missing_refs and not allow_missing: 

964 num_missing = len(missing_refs) 

965 raise FileNotFoundError( 

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

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

968 ) 

969 return uris 

970 

971 @abstractmethod 

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

973 """Return URIs associated with dataset. 

974 

975 Parameters 

976 ---------- 

977 datasetRef : `DatasetRef` 

978 Reference to the required dataset. 

979 predict : `bool`, optional 

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

981 it should return a predicted URI or not. 

982 

983 Returns 

984 ------- 

985 uris : `DatasetRefURIs` 

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

987 the dataset was disassembled within the datastore this may be 

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

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

990 """ 

991 raise NotImplementedError() 

992 

993 @abstractmethod 

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

995 """URI to the Dataset. 

996 

997 Parameters 

998 ---------- 

999 datasetRef : `DatasetRef` 

1000 Reference to the required Dataset. 

1001 predict : `bool` 

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

1003 not exist in datastore. 

1004 

1005 Returns 

1006 ------- 

1007 uri : `str` 

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

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

1010 If the datastore does not have entities that relate well 

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

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

1013 

1014 Raises 

1015 ------ 

1016 FileNotFoundError 

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

1018 guessing is not allowed. 

1019 """ 

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

1021 

1022 @abstractmethod 

1023 def retrieveArtifacts( 

1024 self, 

1025 refs: Iterable[DatasetRef], 

1026 destination: ResourcePath, 

1027 transfer: str = "auto", 

1028 preserve_path: bool = True, 

1029 overwrite: bool = False, 

1030 write_index: bool = True, 

1031 add_prefix: bool = False, 

1032 ) -> dict[ResourcePath, ArtifactIndexInfo]: 

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

1034 

1035 Parameters 

1036 ---------- 

1037 refs : iterable of `DatasetRef` 

1038 The datasets for which artifacts are to be retrieved. 

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

1040 be resolved. 

1041 destination : `lsst.resources.ResourcePath` 

1042 Location to write the artifacts. 

1043 transfer : `str`, optional 

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

1045 supported by `lsst.resources.ResourcePath.transfer_from()`. 

1046 "move" is not allowed. 

1047 preserve_path : `bool`, optional 

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

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

1050 is used. 

1051 overwrite : `bool`, optional 

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

1053 destination. 

1054 write_index : `bool`, optional 

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

1056 of a `ZipIndex` for the downloaded datasets. 

1057 add_prefix : `bool`, optional 

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

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

1060 This can be used to guarantee uniqueness. 

1061 

1062 Returns 

1063 ------- 

1064 artifact_map : `dict` [ `lsst.resources.ResourcePath`, \ 

1065 `ArtifactIndexInfo` ] 

1066 Mapping of retrieved file to associated index information. 

1067 

1068 Notes 

1069 -----