Coverage for python/lsst/daf/butler/registry/sql_registry.py: 22%
405 statements
« prev ^ index » next coverage.py v7.11.0, created at 2025-11-13 01:31 -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/>.
28from __future__ import annotations
30from .. import ddl
32__all__ = ("SqlRegistry",)
34import contextlib
35import logging
36import warnings
37from collections import defaultdict
38from collections.abc import Iterable, Iterator, Mapping, Sequence
39from typing import TYPE_CHECKING, Any
41import sqlalchemy
43from lsst.resources import ResourcePathExpression
44from lsst.utils.iteration import ensure_iterable
46from .._collection_type import CollectionType
47from .._config import Config
48from .._dataset_ref import DatasetId, DatasetIdGenEnum, DatasetRef
49from .._dataset_type import DatasetType
50from .._exceptions import DataIdValueError, DimensionNameError, InconsistentDataIdError
51from .._storage_class import StorageClassFactory
52from .._timespan import Timespan
53from ..dimensions import (
54 DataCoordinate,
55 DataId,
56 DimensionConfig,
57 DimensionDataAttacher,
58 DimensionElement,
59 DimensionGroup,
60 DimensionRecord,
61 DimensionUniverse,
62)
63from ..dimensions.record_cache import DimensionRecordCache
64from ..direct_query_driver import DirectQueryDriver
65from ..progress import Progress
66from ..queries import Query
67from ..registry import (
68 CollectionExpressionError,
69 CollectionSummary,
70 CollectionTypeError,
71 ConflictingDefinitionError,
72 NoDefaultCollectionError,
73 OrphanedRecordError,
74 RegistryConfig,
75 RegistryDefaults,
76)
77from ..registry.interfaces import ChainedCollectionRecord, ReadOnlyDatabaseError, RunRecord
78from ..registry.managers import RegistryManagerInstances, RegistryManagerTypes
79from ..registry.wildcards import CollectionWildcard, DatasetTypeWildcard
80from ..utils import transactional
82if TYPE_CHECKING:
83 from .._butler_config import ButlerConfig
84 from ..datastore._datastore import DatastoreOpaqueTable
85 from ..datastore.stored_file_info import StoredDatastoreItemInfo
86 from ..registry.interfaces import (
87 CollectionRecord,
88 Database,
89 DatastoreRegistryBridgeManager,
90 ObsCoreTableManager,
91 )
94_LOG = logging.getLogger(__name__)
97class SqlRegistry:
98 """Butler Registry implementation that uses SQL database as backend.
100 Parameters
101 ----------
102 database : `Database`
103 Database instance to store Registry.
104 defaults : `RegistryDefaults`
105 Default collection search path and/or output `~CollectionType.RUN`
106 collection.
107 managers : `RegistryManagerInstances`
108 All the managers required for this registry.
109 """
111 defaultConfigFile: str | None = None
112 """Path to configuration defaults. Accessed within the ``configs`` resource
113 or relative to a search path. Can be None if no defaults specified.
114 """
116 @classmethod
117 def forceRegistryConfig(
118 cls, config: ButlerConfig | RegistryConfig | Config | str | None
119 ) -> RegistryConfig:
120 """Force the supplied config to a `RegistryConfig`.
122 Parameters
123 ----------
124 config : `RegistryConfig`, `Config` or `str` or `None`
125 Registry configuration, if missing then default configuration will
126 be loaded from registry.yaml.
128 Returns
129 -------
130 registry_config : `RegistryConfig`
131 A registry config.
132 """
133 if not isinstance(config, RegistryConfig):
134 if isinstance(config, str | Config) or config is None:
135 config = RegistryConfig(config)
136 else:
137 raise ValueError(f"Incompatible Registry configuration: {config}")
138 return config
140 @classmethod
141 def createFromConfig(
142 cls,
143 config: RegistryConfig | str | None = None,
144 dimensionConfig: DimensionConfig | str | None = None,
145 butlerRoot: ResourcePathExpression | None = None,
146 ) -> SqlRegistry:
147 """Create registry database and return `SqlRegistry` instance.
149 This method initializes database contents, database must be empty
150 prior to calling this method.
152 Parameters
153 ----------
154 config : `RegistryConfig` or `str`, optional
155 Registry configuration, if missing then default configuration will
156 be loaded from registry.yaml.
157 dimensionConfig : `DimensionConfig` or `str`, optional
158 Dimensions configuration, if missing then default configuration
159 will be loaded from dimensions.yaml.
160 butlerRoot : convertible to `lsst.resources.ResourcePath`, optional
161 Path to the repository root this `SqlRegistry` will manage.
163 Returns
164 -------
165 registry : `SqlRegistry`
166 A new `SqlRegistry` instance.
167 """
168 config = cls.forceRegistryConfig(config)
169 config.replaceRoot(butlerRoot)
171 if isinstance(dimensionConfig, str):
172 dimensionConfig = DimensionConfig(dimensionConfig)
173 elif dimensionConfig is None:
174 dimensionConfig = DimensionConfig()
175 elif not isinstance(dimensionConfig, DimensionConfig):
176 raise TypeError(f"Incompatible Dimension configuration type: {type(dimensionConfig)}")
178 DatabaseClass = config.getDatabaseClass()
179 database = DatabaseClass.fromUri(
180 config.connectionString,
181 origin=config.get("origin", 0),
182 namespace=config.get("namespace"),
183 allow_temporary_tables=config.areTemporaryTablesAllowed,
184 )
185 managerTypes = RegistryManagerTypes.fromConfig(config)
186 managers = managerTypes.makeRepo(database, dimensionConfig)
187 return cls(database, RegistryDefaults(), managers)
189 @classmethod
190 def fromConfig(
191 cls,
192 config: ButlerConfig | RegistryConfig | Config | str,
193 butlerRoot: ResourcePathExpression | None = None,
194 writeable: bool = True,
195 defaults: RegistryDefaults | None = None,
196 ) -> SqlRegistry:
197 """Create `Registry` subclass instance from `config`.
199 Registry database must be initialized prior to calling this method.
201 Parameters
202 ----------
203 config : `ButlerConfig`, `RegistryConfig`, `Config` or `str`
204 Registry configuration.
205 butlerRoot : `lsst.resources.ResourcePathExpression`, optional
206 Path to the repository root this `Registry` will manage.
207 writeable : `bool`, optional
208 If `True` (default) create a read-write connection to the database.
209 defaults : `RegistryDefaults`, optional
210 Default collection search path and/or output `~CollectionType.RUN`
211 collection.
213 Returns
214 -------
215 registry : `SqlRegistry`
216 A new `SqlRegistry` subclass instance.
217 """
218 config = cls.forceRegistryConfig(config)
219 config.replaceRoot(butlerRoot)
220 DatabaseClass = config.getDatabaseClass()
221 database = DatabaseClass.fromUri(
222 config.connectionString,
223 origin=config.get("origin", 0),
224 namespace=config.get("namespace"),
225 writeable=writeable,
226 allow_temporary_tables=config.areTemporaryTablesAllowed,
227 )
228 managerTypes = RegistryManagerTypes.fromConfig(config)
229 with database.session():
230 managers = managerTypes.loadRepo(database)
231 if defaults is None:
232 defaults = RegistryDefaults()
234 return cls(database, defaults, managers)
236 def __init__(
237 self,
238 database: Database,
239 defaults: RegistryDefaults,
240 managers: RegistryManagerInstances,
241 ):
242 self._db = database
243 self._managers = managers
244 if managers.obscore is not None:
245 managers.obscore.set_query_function(self._query)
246 self.storageClasses = StorageClassFactory()
247 # This is public to SqlRegistry's internal-to-daf_butler callers, but
248 # it is intentionally not part of RegistryShim.
249 self.dimension_record_cache = DimensionRecordCache(
250 self._managers.dimensions.universe,
251 fetch=self._managers.dimensions.fetch_cache_dict,
252 )
253 # Intentionally invoke property setter to initialize defaults. This
254 # can only be done after most of the rest of Registry has already been
255 # initialized, and must be done before the property getter is used.
256 self.defaults = defaults
257 # TODO: This is currently initialized by `make_datastore_tables`,
258 # eventually we'll need to do it during construction.
259 # The mapping is indexed by the opaque table name.
260 self._datastore_record_classes: Mapping[str, type[StoredDatastoreItemInfo]] = {}
262 def __str__(self) -> str:
263 return str(self._db)
265 def __repr__(self) -> str:
266 return f"SqlRegistry({self._db!r}, {self.dimensions!r})"
268 def isWriteable(self) -> bool:
269 """Return `True` if this registry allows write operations, and `False`
270 otherwise.
271 """
272 return self._db.isWriteable()
274 def copy(self, defaults: RegistryDefaults | None = None) -> SqlRegistry:
275 """Create a new `SqlRegistry` backed by the same data repository
276 as this one and sharing a database connection pool with it, but with
277 independent defaults and database sessions.
279 Parameters
280 ----------
281 defaults : `~lsst.daf.butler.registry.RegistryDefaults`, optional
282 Default collections and data ID values for the new registry. If
283 not provided, ``self.defaults`` will be used (but future changes
284 to either registry's defaults will not affect the other).
286 Returns
287 -------
288 copy : `SqlRegistry`
289 A new `SqlRegistry` instance with its own defaults.
290 """
291 if defaults is None:
292 # No need to copy, because `RegistryDefaults` is immutable; we
293 # effectively copy on write.
294 defaults = self.defaults
295 db = self._db.clone()
296 result = SqlRegistry(db, defaults, self._managers.clone(db))
297 result._datastore_record_classes = dict(self._datastore_record_classes)
298 result.dimension_record_cache.load_from(self.dimension_record_cache)
299 return result
301 @property
302 def dimensions(self) -> DimensionUniverse:
303 """Definitions of all dimensions recognized by this `Registry`
304 (`DimensionUniverse`).
305 """
306 return self._managers.dimensions.universe
308 @property
309 def defaults(self) -> RegistryDefaults:
310 """Default collection search path and/or output `~CollectionType.RUN`
311 collection (`~lsst.daf.butler.registry.RegistryDefaults`).
313 This is an immutable struct whose components may not be set
314 individually, but the entire struct can be set by assigning to this
315 property.
316 """
317 return self._defaults
319 @defaults.setter
320 def defaults(self, value: RegistryDefaults) -> None:
321 if value.run is not None:
322 self.registerRun(value.run)
323 value.finish(self)
324 self._defaults = value
326 def refresh(self) -> None:
327 """Refresh all in-memory state by querying the database.
329 This may be necessary to enable querying for entities added by other
330 registry instances after this one was constructed.
331 """
332 self.dimension_record_cache.reset()
333 with self._db.transaction():
334 self._managers.refresh()
336 def refresh_collection_summaries(self) -> None:
337 """Refresh content of the collection summary tables in the database.
339 This only cleans dataset type summaries, we may want to add cleanup of
340 governor summaries later.
341 """
342 for dataset_type in self.queryDatasetTypes():
343 self._managers.datasets.refresh_collection_summaries(dataset_type)
345 def caching_context(self) -> contextlib.AbstractContextManager[None]:
346 """Return context manager that enables caching.
348 Returns
349 -------
350 manager
351 A context manager that enables client-side caching. Entering
352 the context returns `None`.
353 """
354 return self._managers.caching_context_manager()
356 @contextlib.contextmanager
357 def transaction(self, *, savepoint: bool = False) -> Iterator[None]:
358 """Return a context manager that represents a transaction.
360 Parameters
361 ----------
362 savepoint : `bool`
363 Whether to issue a SAVEPOINT in the database.
365 Yields
366 ------
367 `None`
368 """
369 with self._db.transaction(savepoint=savepoint):
370 yield
372 def resetConnectionPool(self) -> None:
373 """Reset SQLAlchemy connection pool for `SqlRegistry` database.
375 This operation is useful when using registry with fork-based
376 multiprocessing. To use registry across fork boundary one has to make
377 sure that there are no currently active connections (no session or
378 transaction is in progress) and connection pool is reset using this
379 method. This method should be called by the child process immediately
380 after the fork.
381 """
382 self._db._engine.dispose()
384 def registerOpaqueTable(self, tableName: str, spec: ddl.TableSpec) -> None:
385 """Add an opaque (to the `Registry`) table for use by a `Datastore` or
386 other data repository client.
388 Opaque table records can be added via `insertOpaqueData`, retrieved via
389 `fetchOpaqueData`, and removed via `deleteOpaqueData`.
391 Parameters
392 ----------
393 tableName : `str`
394 Logical name of the opaque table. This may differ from the
395 actual name used in the database by a prefix and/or suffix.
396 spec : `ddl.TableSpec`
397 Specification for the table to be added.
398 """
399 self._managers.opaque.register(tableName, spec)
401 @transactional
402 def insertOpaqueData(self, tableName: str, *data: dict) -> None:
403 """Insert records into an opaque table.
405 Parameters
406 ----------
407 tableName : `str`
408 Logical name of the opaque table. Must match the name used in a
409 previous call to `registerOpaqueTable`.
410 *data
411 Each additional positional argument is a dictionary that represents
412 a single row to be added.
413 """
414 self._managers.opaque[tableName].insert(*data)
416 def fetchOpaqueData(self, tableName: str, **where: Any) -> Iterator[Mapping[str, Any]]:
417 """Retrieve records from an opaque table.
419 Parameters
420 ----------
421 tableName : `str`
422 Logical name of the opaque table. Must match the name used in a
423 previous call to `registerOpaqueTable`.
424 **where
425 Additional keyword arguments are interpreted as equality
426 constraints that restrict the returned rows (combined with AND);
427 keyword arguments are column names and values are the values they
428 must have.
430 Yields
431 ------
432 row : `dict`
433 A dictionary representing a single result row.
434 """
435 yield from self._managers.opaque[tableName].fetch(**where)
437 @transactional
438 def deleteOpaqueData(self, tableName: str, **where: Any) -> None:
439 """Remove records from an opaque table.
441 Parameters
442 ----------
443 tableName : `str`
444 Logical name of the opaque table. Must match the name used in a
445 previous call to `registerOpaqueTable`.
446 **where
447 Additional keyword arguments are interpreted as equality
448 constraints that restrict the deleted rows (combined with AND);
449 keyword arguments are column names and values are the values they
450 must have.
451 """
452 self._managers.opaque[tableName].delete(where.keys(), where)
454 def registerCollection(
455 self, name: str, type: CollectionType = CollectionType.TAGGED, doc: str | None = None
456 ) -> bool:
457 """Add a new collection if one with the given name does not exist.
459 Parameters
460 ----------
461 name : `str`
462 The name of the collection to create.
463 type : `CollectionType`
464 Enum value indicating the type of collection to create.
465 doc : `str`, optional
466 Documentation string for the collection.
468 Returns
469 -------
470 registered : `bool`
471 Boolean indicating whether the collection was already registered
472 or was created by this call.
474 Notes
475 -----
476 This method cannot be called within transactions, as it needs to be
477 able to perform its own transaction to be concurrent.
478 """
479 _, registered = self._managers.collections.register(name, type, doc=doc)
480 return registered
482 def getCollectionType(self, name: str) -> CollectionType:
483 """Return an enumeration value indicating the type of the given
484 collection.
486 Parameters
487 ----------
488 name : `str`
489 The name of the collection.
491 Returns
492 -------
493 type : `CollectionType`
494 Enum value indicating the type of this collection.
496 Raises
497 ------
498 lsst.daf.butler.registry.MissingCollectionError
499 Raised if no collection with the given name exists.
500 """
501 return self._managers.collections.find(name).type
503 def get_collection_record(self, name: str) -> CollectionRecord:
504 """Return the record for this collection.
506 Parameters
507 ----------
508 name : `str`
509 Name of the collection for which the record is to be retrieved.
511 Returns
512 -------
513 record : `CollectionRecord`
514 The record for this collection.
515 """
516 return self._managers.collections.find(name)
518 def registerRun(self, name: str, doc: str | None = None) -> bool:
519 """Add a new run if one with the given name does not exist.
521 Parameters
522 ----------
523 name : `str`
524 The name of the run to create.
525 doc : `str`, optional
526 Documentation string for the collection.
528 Returns
529 -------
530 registered : `bool`
531 Boolean indicating whether a new run was registered. `False`
532 if it already existed.
534 Notes
535 -----
536 This method cannot be called within transactions, as it needs to be
537 able to perform its own transaction to be concurrent.
538 """
539 _, registered = self._managers.collections.register(name, CollectionType.RUN, doc=doc)
540 return registered
542 @transactional
543 def removeCollection(self, name: str) -> None:
544 """Remove the given collection from the registry.
546 Parameters
547 ----------
548 name : `str`
549 The name of the collection to remove.
551 Raises
552 ------
553 lsst.daf.butler.registry.MissingCollectionError
554 Raised if no collection with the given name exists.
555 sqlalchemy.exc.IntegrityError
556 Raised if the database rows associated with the collection are
557 still referenced by some other table, such as a dataset in a
558 datastore (for `~CollectionType.RUN` collections only) or a
559 `~CollectionType.CHAINED` collection of which this collection is
560 a child.
562 Notes
563 -----
564 If this is a `~CollectionType.RUN` collection, all datasets and quanta
565 in it will removed from the `Registry` database. This requires that
566 those datasets be removed (or at least trashed) from any datastores
567 that hold them first.
569 A collection may not be deleted as long as it is referenced by a
570 `~CollectionType.CHAINED` collection; the ``CHAINED`` collection must
571 be deleted or redefined first.
572 """
573 self._managers.collections.remove(name)
575 def getCollectionChain(self, parent: str) -> tuple[str, ...]:
576 """Return the child collections in a `~CollectionType.CHAINED`
577 collection.
579 Parameters
580 ----------
581 parent : `str`
582 Name of the chained collection. Must have already been added via
583 a call to `Registry.registerCollection`.
585 Returns
586 -------
587 children : `~collections.abc.Sequence` [ `str` ]
588 An ordered sequence of collection names that are searched when the
589 given chained collection is searched.
591 Raises
592 ------
593 lsst.daf.butler.registry.MissingCollectionError
594 Raised if ``parent`` does not exist in the `Registry`.
595 lsst.daf.butler.registry.CollectionTypeError
596 Raised if ``parent`` does not correspond to a
597 `~CollectionType.CHAINED` collection.
598 """
599 record = self._managers.collections.find(parent)
600 if record.type is not CollectionType.CHAINED:
601 raise CollectionTypeError(f"Collection '{parent}' has type {record.type.name}, not CHAINED.")
602 assert isinstance(record, ChainedCollectionRecord)
603 return record.children
605 @transactional
606 def setCollectionChain(self, parent: str, children: Any, *, flatten: bool = False) -> None:
607 """Define or redefine a `~CollectionType.CHAINED` collection.
609 Parameters
610 ----------
611 parent : `str`
612 Name of the chained collection. Must have already been added via
613 a call to `Registry.registerCollection`.
614 children : collection expression
615 An expression defining an ordered search of child collections,
616 generally an iterable of `str`; see
617 :ref:`daf_butler_collection_expressions` for more information.
618 flatten : `bool`, optional
619 If `True` (`False` is default), recursively flatten out any nested
620 `~CollectionType.CHAINED` collections in ``children`` first.
622 Raises
623 ------
624 lsst.daf.butler.registry.MissingCollectionError
625 Raised when any of the given collections do not exist in the
626 `Registry`.
627 lsst.daf.butler.registry.CollectionTypeError
628 Raised if ``parent`` does not correspond to a
629 `~CollectionType.CHAINED` collection.
630 CollectionCycleError
631 Raised if the given collections contains a cycle.
633 Notes
634 -----
635 If this function is called within a call to ``Butler.transaction``, it
636 will hold a lock that prevents other processes from modifying the
637 parent collection until the end of the transaction. Keep these
638 transactions short.
639 """
640 children = CollectionWildcard.from_expression(children).require_ordered()
641 if flatten:
642 children = self.queryCollections(children, flattenChains=True)
644 self._managers.collections.update_chain(parent, list(children), allow_use_in_caching_context=True)
646 def getCollectionParentChains(self, collection: str) -> set[str]:
647 """Return the CHAINED collections that directly contain the given one.
649 Parameters
650 ----------
651 collection : `str`
652 Name of the collection.
654 Returns
655 -------
656 chains : `set` of `str`
657 Set of `~CollectionType.CHAINED` collection names.
658 """
659 return self._managers.collections.getParentChains(self._managers.collections.find(collection).key)
661 def getCollectionDocumentation(self, collection: str) -> str | None:
662 """Retrieve the documentation string for a collection.
664 Parameters
665 ----------
666 collection : `str`
667 Name of the collection.
669 Returns
670 -------
671 docs : `str` or `None`
672 Docstring for the collection with the given name.
673 """
674 return self._managers.collections.getDocumentation(self._managers.collections.find(collection).key)
676 def setCollectionDocumentation(self, collection: str, doc: str | None) -> None:
677 """Set the documentation string for a collection.
679 Parameters
680 ----------
681 collection : `str`
682 Name of the collection.
683 doc : `str` or `None`
684 Docstring for the collection with the given name; will replace any
685 existing docstring. Passing `None` will remove any existing
686 docstring.
687 """
688 self._managers.collections.setDocumentation(self._managers.collections.find(collection).key, doc)
690 def getCollectionSummary(self, collection: str) -> CollectionSummary:
691 """Return a summary for the given collection.
693 Parameters
694 ----------
695 collection : `str`
696 Name of the collection for which a summary is to be retrieved.
698 Returns
699 -------
700 summary : `~lsst.daf.butler.registry.CollectionSummary`
701 Summary of the dataset types and governor dimension values in
702 this collection.
703 """
704 record = self._managers.collections.find(collection)
705 return self._managers.datasets.getCollectionSummary(record)
707 def registerDatasetType(self, datasetType: DatasetType) -> bool:
708 """Add a new `DatasetType` to the Registry.
710 It is not an error to register the same `DatasetType` twice.
712 Parameters
713 ----------
714 datasetType : `DatasetType`
715 The `DatasetType` to be added.
717 Returns
718 -------
719 inserted : `bool`
720 `True` if ``datasetType`` was inserted, `False` if an identical
721 existing `DatasetType` was found. Note that in either case the
722 DatasetType is guaranteed to be defined in the Registry
723 consistently with the given definition.
725 Raises
726 ------
727 ValueError
728 Raised if the dimensions or storage class are invalid.
729 lsst.daf.butler.registry.ConflictingDefinitionError
730 Raised if this `DatasetType` is already registered with a different
731 definition.
733 Notes
734 -----
735 This method cannot be called within transactions, as it needs to be
736 able to perform its own transaction to be concurrent.
737 """
738 return self._managers.datasets.register_dataset_type(datasetType)
740 def removeDatasetType(self, name: str | tuple[str, ...]) -> None:
741 """Remove the named `DatasetType` from the registry.
743 .. warning::
745 Registry implementations can cache the dataset type definitions.
746 This means that deleting the dataset type definition may result in
747 unexpected behavior from other butler processes that are active
748 that have not seen the deletion.
750 Parameters
751 ----------
752 name : `str` or `tuple` [`str`]
753 Name of the type to be removed or tuple containing a list of type
754 names to be removed. Wildcards are allowed.
756 Raises
757 ------
758 lsst.daf.butler.registry.OrphanedRecordError
759 Raised if an attempt is made to remove the dataset type definition
760 when there are already datasets associated with it.
762 Notes
763 -----
764 If the dataset type is not registered the method will return without
765 action.
766 """
767 for datasetTypeExpression in ensure_iterable(name):
768 # Catch any warnings from the caller specifying a component
769 # dataset type. This will result in an error later but the
770 # warning could be confusing when the caller is not querying
771 # anything.
772 with warnings.catch_warnings():
773 warnings.simplefilter("ignore", category=FutureWarning)
774 datasetTypes = list(self.queryDatasetTypes(datasetTypeExpression))
775 if not datasetTypes:
776 _LOG.info("Dataset type %r not defined", datasetTypeExpression)
777 else:
778 for datasetType in datasetTypes:
779 self._managers.datasets.remove_dataset_type(datasetType.name)
780 _LOG.info("Removed dataset type %r", datasetType.name)
782 def getDatasetType(self, name: str) -> DatasetType:
783 """Get the `DatasetType`.
785 Parameters
786 ----------
787 name : `str`
788 Name of the type.
790 Returns
791 -------
792 type : `DatasetType`
793 The `DatasetType` associated with the given name.
795 Raises
796 ------
797 lsst.daf.butler.registry.MissingDatasetTypeError
798 Raised if the requested dataset type has not been registered.
800 Notes
801 -----
802 This method handles component dataset types automatically, though most
803 other registry operations do not.
804 """
805 parent_name, component = DatasetType.splitDatasetTypeName(name)
806 parent_dataset_type = self._managers.datasets.get_dataset_type(parent_name)
807 if component is None:
808 return parent_dataset_type
809 else:
810 return parent_dataset_type.makeComponentDatasetType(component)
812 def supportsIdGenerationMode(self, mode: DatasetIdGenEnum) -> bool:
813 """Test whether the given dataset ID generation mode is supported by
814 `insertDatasets`.
816 Parameters
817 ----------
818 mode : `DatasetIdGenEnum`
819 Enum value for the mode to test.
821 Returns
822 -------
823 supported : `bool`
824 Whether the given mode is supported.
825 """
826 return True
828 @transactional
829 def insertDatasets(
830 self,
831 datasetType: DatasetType | str,
832 dataIds: Iterable[DataId],
833 run: str | None = None,
834 expand: bool = True,
835 idGenerationMode: DatasetIdGenEnum = DatasetIdGenEnum.UNIQUE,
836 ) -> list[DatasetRef]:
837 """Insert one or more datasets into the `Registry`.
839 This always adds new datasets; to associate existing datasets with
840 a new collection, use ``associate``.
842 Parameters
843 ----------
844 datasetType : `DatasetType` or `str`
845 A `DatasetType` or the name of one.
846 dataIds : `~collections.abc.Iterable` of `dict` or `DataCoordinate`
847 Dimension-based identifiers for the new datasets.
848 run : `str`, optional
849 The name of the run that produced the datasets. Defaults to
850 ``self.defaults.run``.
851 expand : `bool`, optional
852 If `True` (default), expand data IDs as they are inserted. This is
853 necessary in general to allow datastore to generate file templates,
854 but it may be disabled if the caller can guarantee this is
855 unnecessary.
856 idGenerationMode : `DatasetIdGenEnum`, optional
857 Specifies option for generating dataset IDs. By default unique IDs
858 are generated for each inserted dataset.
860 Returns
861 -------
862 refs : `list` of `DatasetRef`
863 Resolved `DatasetRef` instances for all given data IDs (in the same
864 order).
866 Raises
867 ------
868 lsst.daf.butler.registry.DatasetTypeError
869 Raised if ``datasetType`` is not known to registry.
870 lsst.daf.butler.registry.CollectionTypeError
871 Raised if ``run`` collection type is not `~CollectionType.RUN`.
872 lsst.daf.butler.registry.NoDefaultCollectionError
873 Raised if ``run`` is `None` and ``self.defaults.run`` is `None`.
874 lsst.daf.butler.registry.ConflictingDefinitionError
875 If a dataset with the same dataset type and data ID as one of those
876 given already exists in ``run``.
877 lsst.daf.butler.registry.MissingCollectionError
878 Raised if ``run`` does not exist in the registry.
879 """
880 datasetType = self._managers.datasets.conform_exact_dataset_type(datasetType)
881 if run is None:
882 if self.defaults.run is None:
883 raise NoDefaultCollectionError(
884 "No run provided to insertDatasets, and no default from registry construction."
885 )
886 run = self.defaults.run
887 runRecord = self._managers.collections.find(run)
888 if runRecord.type is not CollectionType.RUN:
889 raise CollectionTypeError(
890 f"Given collection is of type {runRecord.type.name}; RUN collection required."
891 )
892 assert isinstance(runRecord, RunRecord)
894 expandedDataIds = [
895 DataCoordinate.standardize(dataId, dimensions=datasetType.dimensions) for dataId in dataIds
896 ]
897 if expand:
898 _LOG.debug("Expanding %d data IDs", len(expandedDataIds))
899 expandedDataIds = self.expand_data_ids(expandedDataIds)
900 _LOG.debug("Finished expanding data IDs")
902 try:
903 refs = list(
904 self._managers.datasets.insert(datasetType.name, runRecord, expandedDataIds, idGenerationMode)
905 )
906 if self._managers.obscore:
907 self._managers.obscore.add_datasets(refs)
908 except sqlalchemy.exc.IntegrityError as err:
909 raise ConflictingDefinitionError(
910 "A database constraint failure was triggered by inserting "
911 f"one or more datasets of type {datasetType} into "
912 f"collection '{run}'. "
913 "This probably means a dataset with the same data ID "
914 "and dataset type already exists, but it may also mean a "
915 "dimension row is missing."
916 ) from err
917 return refs
919 @transactional
920 def _importDatasets(
921 self,
922 datasets: Iterable[DatasetRef],
923 expand: bool = True,
924 assume_new: bool = False,
925 ) -> list[DatasetRef]:
926 """Import one or more datasets into the `Registry`.
928 This differs from `insertDatasets` method in that this method accepts
929 `DatasetRef` instances, which already have a dataset ID.
931 Parameters
932 ----------
933 datasets : `~collections.abc.Iterable` of `DatasetRef`
934 Datasets to be inserted. All `DatasetRef` instances must have
935 identical ``run`` attributes. ``run``
936 attribute can be `None` and defaults to ``self.defaults.run``.
937 Datasets can specify ``id`` attribute which will be used for
938 inserted datasets.
939 Datasets can be of multiple dataset types, but all the dataset
940 types must have the same set of dimensions.
941 expand : `bool`, optional
942 If `True` (default), expand data IDs as they are inserted. This is
943 necessary in general, but it may be disabled if the caller can
944 guarantee this is unnecessary.
945 assume_new : `bool`, optional
946 If `True`, assume datasets are new. If `False`, datasets that are
947 identical to an existing one are ignored.
949 Returns
950 -------
951 refs : `list` of `DatasetRef`
952 `DatasetRef` instances for all given data IDs (in the same order).
953 If any of ``datasets`` has an ID which already exists in the
954 database then it will not be inserted or updated, but a
955 `DatasetRef` will be returned for it in any case.
957 Raises
958 ------
959 lsst.daf.butler.registry.NoDefaultCollectionError
960 Raised if ``run`` is `None` and ``self.defaults.run`` is `None`.
961 lsst.daf.butler.registry.DatasetTypeError
962 Raised if a dataset type is not known to registry.
963 lsst.daf.butler.registry.ConflictingDefinitionError
964 If a dataset with the same dataset type and data ID as one of those
965 given already exists in ``run``, or if ``assume_new=True`` and at
966 least one dataset is not new.
967 lsst.daf.butler.registry.MissingCollectionError
968 Raised if ``run`` does not exist in the registry.
970 Notes
971 -----
972 This method is considered middleware-internal.
973 """
974 datasets = list(datasets)
975 if not datasets:
976 # nothing to do
977 return []
979 # find run name
980 runs = {dataset.run for dataset in datasets}
981 if len(runs) != 1:
982 raise ValueError(f"Multiple run names in input datasets: {runs}")
983 run = runs.pop()
985 runRecord = self._managers.collections.find(run)
986 if runRecord.type is not CollectionType.RUN:
987 raise CollectionTypeError(
988 f"Given collection '{runRecord.name}' is of type {runRecord.type.name};"
989 " RUN collection required."
990 )
991 assert isinstance(runRecord, RunRecord)
993 if expand:
994 _LOG.debug("Expanding %d data IDs", len(datasets))
995 datasets = self.expand_refs(datasets)
996 _LOG.debug("Finished expanding data IDs")
998 try:
999 self._managers.datasets.import_(runRecord, datasets, assume_new=assume_new)
1000 if self._managers.obscore:
1001 self._managers.obscore.add_datasets(datasets)
1002 except sqlalchemy.exc.IntegrityError as err:
1003 raise ConflictingDefinitionError(
1004 "A database constraint failure was triggered by inserting "
1005 f"one or more datasets into collection '{run}'. "
1006 "This probably means a dataset with the same data ID "
1007 "and dataset type already exists, but it may also mean a "
1008 "dimension row is missing, or the dataset was assumed to be "
1009 "new when it was not."
1010 ) from err
1011 return datasets
1013 def getDataset(self, id: DatasetId) -> DatasetRef | None:
1014 """Retrieve a Dataset entry.
1016 Parameters
1017 ----------
1018 id : `DatasetId`
1019 The unique identifier for the dataset.
1021 Returns
1022 -------
1023 ref : `DatasetRef` or `None`
1024 A ref to the Dataset, or `None` if no matching Dataset
1025 was found.
1026 """
1027 return self._managers.datasets.getDatasetRef(id)
1029 def _fetch_run_dataset_ids(self, run: str) -> list[DatasetId]:
1030 """Return the IDs of all datasets in the given ``RUN``
1031 collection.
1033 Parameters
1034 ----------
1035 run : `str`
1036 Name of the collection.
1038 Returns
1039 -------
1040 dataset_ids : `list` [`uuid.UUID`]
1041 List of dataset IDs.
1043 Notes
1044 -----
1045 This is a middleware-internal interface.
1046 """
1047 run_record = self._managers.collections.find(run)
1048 if not isinstance(run_record, RunRecord):
1049 raise CollectionTypeError(f"{run!r} is not a RUN collection.")
1050 return self._managers.datasets.fetch_run_dataset_ids(run_record)
1052 @transactional
1053 def removeDatasets(self, refs: Iterable[DatasetRef]) -> None:
1054 """Remove datasets from the Registry.
1056 The datasets will be removed unconditionally from all collections.
1057 `Datastore` records will *not* be deleted; the caller is responsible
1058 for ensuring that the dataset has already been removed from all
1059 Datastores.
1061 Parameters
1062 ----------
1063 refs : `~collections.abc.Iterable` [`DatasetRef`]
1064 References to the datasets to be removed. Should be considered
1065 invalidated upon return.
1067 Raises
1068 ------
1069 lsst.daf.butler.registry.OrphanedRecordError
1070 Raised if any dataset is still present in any `Datastore`.
1071 """
1072 try:
1073 self._managers.datasets.delete(refs)
1074 except sqlalchemy.exc.IntegrityError as err:
1075 raise OrphanedRecordError(
1076 "One or more datasets is still present in one or more Datastores."
1077 ) from err
1079 @transactional
1080 def associate(self, collection: str, refs: Iterable[DatasetRef]) -> None:
1081 """Add existing datasets to a `~CollectionType.TAGGED` collection.
1083 If a DatasetRef with the same exact ID is already in a collection
1084 nothing is changed. If a `DatasetRef` with the same `DatasetType` and
1085 data ID but with different ID exists in the collection,
1086 `~lsst.daf.butler.registry.ConflictingDefinitionError` is raised.
1088 Parameters
1089 ----------
1090 collection : `str`
1091 Indicates the collection the datasets should be associated with.
1092 refs : `~collections.abc.Iterable` [ `DatasetRef` ]
1093 An iterable of resolved `DatasetRef` instances that already exist
1094 in this `Registry`.
1096 Raises
1097 ------
1098 lsst.daf.butler.registry.ConflictingDefinitionError
1099 If a Dataset with the given `DatasetRef` already exists in the
1100 given collection.
1101 lsst.daf.butler.registry.MissingCollectionError
1102 Raised if ``collection`` does not exist in the registry.
1103 lsst.daf.butler.registry.CollectionTypeError
1104 Raise adding new datasets to the given ``collection`` is not
1105 allowed.
1106 """
1107 progress = Progress("lsst.daf.butler.Registry.associate", level=logging.DEBUG)
1108 collectionRecord = self._managers.collections.find(collection)
1109 for datasetType, refsForType in progress.iter_item_chunks(
1110 DatasetRef.iter_by_type(refs), desc="Associating datasets by type"
1111 ):
1112 try:
1113 self._managers.datasets.associate(datasetType, collectionRecord, refsForType)
1114 if self._managers.obscore:
1115 # If a TAGGED collection is being monitored by ObsCore
1116 # manager then we may need to save the dataset.
1117 self._managers.obscore.associate(refsForType, collectionRecord)
1118 except sqlalchemy.exc.IntegrityError as err:
1119 raise ConflictingDefinitionError(
1120 f"Constraint violation while associating dataset of type {datasetType.name} with "
1121 f"collection {collection}. This probably means that one or more datasets with the same "
1122 "dataset type and data ID already exist in the collection, but it may also indicate "
1123 "that the datasets do not exist."
1124 ) from err
1126 @transactional
1127 def disassociate(self, collection: str, refs: Iterable[DatasetRef]) -> None:
1128 """Remove existing datasets from a `~CollectionType.TAGGED` collection.
1130 ``collection`` and ``ref`` combinations that are not currently
1131 associated are silently ignored.
1133 Parameters
1134 ----------
1135 collection : `str`
1136 The collection the datasets should no longer be associated with.
1137 refs : `~collections.abc.Iterable` [ `DatasetRef` ]
1138 An iterable of resolved `DatasetRef` instances that already exist
1139 in this `Registry`.
1141 Raises
1142 ------
1143 lsst.daf.butler.AmbiguousDatasetError
1144 Raised if any of the given dataset references is unresolved.
1145 lsst.daf.butler.registry.MissingCollectionError
1146 Raised if ``collection`` does not exist in the registry.
1147 lsst.daf.butler.registry.CollectionTypeError
1148 Raise adding new datasets to the given ``collection`` is not
1149 allowed.
1150 """
1151 progress = Progress("lsst.daf.butler.Registry.disassociate", level=logging.DEBUG)
1152 collectionRecord = self._managers.collections.find(collection)
1153 for datasetType, refsForType in progress.iter_item_chunks(
1154 DatasetRef.iter_by_type(refs), desc="Disassociating datasets by type"
1155 ):
1156 self._managers.datasets.disassociate(datasetType, collectionRecord, refsForType)
1157 if self._managers.obscore:
1158 self._managers.obscore.disassociate(refsForType, collectionRecord)
1160 @transactional
1161 def certify(self, collection: str, refs: Iterable[DatasetRef], timespan: Timespan) -> None:
1162 """Associate one or more datasets with a calibration collection and a
1163 validity range within it.
1165 Parameters
1166 ----------
1167 collection : `str`
1168 The name of an already-registered `~CollectionType.CALIBRATION`
1169 collection.
1170 refs : `~collections.abc.Iterable` [ `DatasetRef` ]
1171 Datasets to be associated.
1172 timespan : `Timespan`
1173 The validity range for these datasets within the collection.
1175 Raises
1176 ------
1177 lsst.daf.butler.AmbiguousDatasetError
1178 Raised if any of the given `DatasetRef` instances is unresolved.
1179 lsst.daf.butler.registry.ConflictingDefinitionError
1180 Raised if the collection already contains a different dataset with
1181 the same `DatasetType` and data ID and an overlapping validity
1182 range.
1183 DatasetTypeError
1184 Raised if ``ref.datasetType.isCalibration() is False`` for any ref.
1185 CollectionTypeError
1186 Raised if
1187 ``collection.type is not CollectionType.CALIBRATION``.
1188 """
1189 progress = Progress("lsst.daf.butler.Registry.certify", level=logging.DEBUG)
1190 with self._managers.caching_context.enable_collection_record_cache():
1191 collectionRecord = self._managers.collections.find(collection)
1192 for datasetType, refsForType in progress.iter_item_chunks(
1193 DatasetRef.iter_by_type(refs), desc="Certifying datasets by type"
1194 ):
1195 self._managers.datasets.certify(
1196 datasetType, collectionRecord, refsForType, timespan, self._query
1197 )
1199 @transactional
1200 def decertify(
1201 self,
1202 collection: str,
1203 datasetType: str | DatasetType,
1204 timespan: Timespan,
1205 *,
1206 dataIds: Iterable[DataId] | None = None,
1207 ) -> None:
1208 """Remove or adjust datasets to clear a validity range within a
1209 calibration collection.
1211 Parameters
1212 ----------
1213 collection : `str`
1214 The name of an already-registered `~CollectionType.CALIBRATION`
1215 collection.
1216 datasetType : `str` or `DatasetType`
1217 Name or `DatasetType` instance for the datasets to be decertified.
1218 timespan : `Timespan`, optional
1219 The validity range to remove datasets from within the collection.
1220 Datasets that overlap this range but are not contained by it will
1221 have their validity ranges adjusted to not overlap it, which may
1222 split a single dataset validity range into two.
1223 dataIds : iterable [`dict` or `DataCoordinate`], optional
1224 Data IDs that should be decertified within the given validity range
1225 If `None`, all data IDs for ``self.datasetType`` will be
1226 decertified.
1228 Raises
1229 ------
1230 DatasetTypeError
1231 Raised if ``datasetType.isCalibration() is False``.
1232 CollectionTypeError
1233 Raised if
1234 ``collection.type is not CollectionType.CALIBRATION``.
1235 """
1236 collectionRecord = self._managers.collections.find(collection)
1237 if isinstance(datasetType, str):
1238 datasetType = self.getDatasetType(datasetType)
1239 standardizedDataIds = None
1240 if dataIds is not None:
1241 standardizedDataIds = [
1242 DataCoordinate.standardize(d, dimensions=datasetType.dimensions) for d in dataIds
1243 ]
1244 self._managers.datasets.decertify(
1245 datasetType, collectionRecord, timespan, data_ids=standardizedDataIds, query_func=self._query
1246 )
1248 def getDatastoreBridgeManager(self) -> DatastoreRegistryBridgeManager:
1249 """Return an object that allows a new `Datastore` instance to
1250 communicate with this `Registry`.
1252 Returns
1253 -------
1254 manager : `~.interfaces.DatastoreRegistryBridgeManager`
1255 Object that mediates communication between this `Registry` and its
1256 associated datastores.
1257 """
1258 return self._managers.datastores
1260 def getDatasetLocations(self, ref: DatasetRef) -> Iterable[str]:
1261 """Retrieve datastore locations for a given dataset.
1263 Parameters
1264 ----------
1265 ref : `DatasetRef`
1266 A reference to the dataset for which to retrieve storage
1267 information.
1269 Returns
1270 -------
1271 datastores : `~collections.abc.Iterable` [ `str` ]
1272 All the matching datastores holding this dataset.
1274 Raises
1275 ------
1276 lsst.daf.butler.AmbiguousDatasetError
1277 Raised if ``ref.id`` is `None`.
1278 """
1279 return self._managers.datastores.findDatastores(ref)
1281 def expandDataId(
1282 self,
1283 dataId: DataId | None = None,
1284 *,
1285 dimensions: Iterable[str] | DimensionGroup | None = None,
1286 records: Mapping[str, DimensionRecord | None] | None = None,
1287 withDefaults: bool = True,
1288 **kwargs: Any,
1289 ) -> DataCoordinate:
1290 """Expand a dimension-based data ID to include additional information.
1292 Parameters
1293 ----------
1294 dataId : `DataCoordinate` or `dict`, optional
1295 Data ID to be expanded; augmented and overridden by ``kwargs``.
1296 dimensions : `~collections.abc.Iterable` [ `str` ], \
1297 `DimensionGroup`, optional
1298 The dimensions to be identified by the new `DataCoordinate`.
1299 If not provided, will be inferred from the keys of ``dataId`` and
1300 ``**kwargs``, and ``universe`` must be provided unless ``dataId``
1301 is already a `DataCoordinate`.
1302 records : `~collections.abc.Mapping` [`str`, `DimensionRecord`], \
1303 optional
1304 Dimension record data to use before querying the database for that
1305 data, keyed by element name.
1306 withDefaults : `bool`, optional
1307 Utilize ``self.defaults.dataId`` to fill in missing governor
1308 dimension key-value pairs. Defaults to `True` (i.e. defaults are
1309 used).
1310 **kwargs
1311 Additional keywords are treated like additional key-value pairs for
1312 ``dataId``, extending and overriding.
1314 Returns
1315 -------
1316 expanded : `DataCoordinate`
1317 A data ID that includes full metadata for all of the dimensions it
1318 identifies, i.e. guarantees that ``expanded.hasRecords()`` and
1319 ``expanded.hasFull()`` both return `True`.
1321 Raises
1322 ------
1323 lsst.daf.butler.registry.DataIdError
1324 Raised when ``dataId`` or keyword arguments specify unknown
1325 dimensions or values, or when a resulting data ID contains
1326 contradictory key-value pairs, according to dimension
1327 relationships.
1329 Notes
1330 -----
1331 This method cannot be relied upon to reject invalid data ID values
1332 for dimensions that do actually not have any record columns. For
1333 efficiency reasons the records for these dimensions (which have only
1334 dimension key values that are given by the caller) may be constructed
1335 directly rather than obtained from the registry database.
1336 """
1337 if not withDefaults:
1338 defaults = None
1339 else:
1340 defaults = self.defaults.dataId
1341 standardized = DataCoordinate.standardize(
1342 dataId,
1343 dimensions=dimensions,
1344 universe=self.dimensions,
1345 defaults=defaults,
1346 **kwargs,
1347 )
1348 if standardized.hasRecords():
1349 return standardized
1350 if records is None:
1351 records = {}
1352 else:
1353 records = dict(records)
1354 if isinstance(dataId, DataCoordinate) and dataId.hasRecords():
1355 for element_name in dataId.dimensions.elements:
1356 records[element_name] = dataId.records[element_name]
1357 keys: dict[str, str | int] = dict(standardized.mapping)
1358 for element_name in standardized.dimensions.lookup_order:
1359 element = self.dimensions[element_name]
1360 record = records.get(element_name, ...) # Use ... to mean not found; None might mean NULL
1361 if record is ...:
1362 if element_name in self.dimensions.dimensions.names and keys.get(element_name) is None:
1363 raise DimensionNameError(f"No value or null value for dimension {element_name}.")
1364 else:
1365 record = self._managers.dimensions.fetch_one(
1366 element_name,
1367 DataCoordinate.standardize(keys, dimensions=element.minimal_group),
1368 self.dimension_record_cache,
1369 )
1370 records[element_name] = record
1371 if record is not None:
1372 for d in element.implied:
1373 value = getattr(record, d.name)
1374 if keys.setdefault(d.name, value) != value:
1375 raise InconsistentDataIdError(
1376 f"Data ID {standardized} has {d.name}={keys[d.name]!r}, "
1377 f"but {element_name} implies {d.name}={value!r}."
1378 )
1379 else:
1380 if element_name in standardized.dimensions.names:
1381 raise DataIdValueError(
1382 f"Could not fetch record for dimension {element.name} via keys {keys}."
1383 )
1384 if element.defines_relationships:
1385 raise InconsistentDataIdError(
1386 f"Could not fetch record for element {element_name} via keys {keys}, "
1387 "but it is marked as defining relationships; this means one or more dimensions are "
1388 "have inconsistent values.",
1389 )
1390 return DataCoordinate.standardize(keys, dimensions=standardized.dimensions).expanded(records=records)
1392 def expand_data_ids(self, data_ids: Iterable[DataCoordinate]) -> list[DataCoordinate]:
1393 output = list(data_ids)
1395 grouped_by_dimensions: defaultdict[DimensionGroup, list[int]] = defaultdict(list)
1396 for i, data_id in enumerate(data_ids):
1397 if not data_id.hasRecords():
1398 grouped_by_dimensions[data_id.dimensions].append(i)
1400 if not grouped_by_dimensions:
1401 # All given DataCoordinate values are already expanded.
1402 return output
1404 attacher = DimensionDataAttacher(
1405 cache=self.dimension_record_cache,
1406 dimensions=DimensionGroup.union(*grouped_by_dimensions.keys(), universe=self.dimensions),
1407 )
1408 with self._query() as query:
1409 for dimensions, indexes in grouped_by_dimensions.items():
1410 expanded = attacher.attach(dimensions, (output[index] for index in indexes), query)
1411 for index, data_id in zip(indexes, expanded):
1412 output[index] = data_id
1414 return output
1416 def expand_refs(self, dataset_refs: list[DatasetRef]) -> list[DatasetRef]:
1417 expanded_ids = self.expand_data_ids([ref.dataId for ref in dataset_refs])
1418 return [ref.expanded(data_id) for ref, data_id in zip(dataset_refs, expanded_ids)]
1420 def insertDimensionData(
1421 self,
1422 element: DimensionElement | str,
1423 *data: Mapping[str, Any] | DimensionRecord,
1424 conform: bool = True,
1425 replace: bool = False,
1426 skip_existing: bool = False,
1427 ) -> None:
1428 """Insert one or more dimension records into the database.
1430 Parameters
1431 ----------
1432 element : `DimensionElement` or `str`
1433 The `DimensionElement` or name thereof that identifies the table
1434 records will be inserted into.
1435 *data : `dict` or `DimensionRecord`
1436 One or more records to insert.
1437 conform : `bool`, optional
1438 If `False` (`True` is default) perform no checking or conversions,
1439 and assume that ``element`` is a `DimensionElement` instance and
1440 ``data`` is a one or more `DimensionRecord` instances of the
1441 appropriate subclass.
1442 replace : `bool`, optional
1443 If `True` (`False` is default), replace existing records in the
1444 database if there is a conflict.
1445 skip_existing : `bool`, optional
1446 If `True` (`False` is default), skip insertion if a record with
1447 the same primary key values already exists. Unlike
1448 `syncDimensionData`, this will not detect when the given record
1449 differs from what is in the database, and should not be used when
1450 this is a concern.
1451 """
1452 if isinstance(element, str):
1453 element = self.dimensions[element]
1454 if conform:
1455 records = [
1456 row if isinstance(row, DimensionRecord) else element.RecordClass(**row) for row in data
1457 ]
1458 else:
1459 # Ignore typing since caller said to trust them with conform=False.
1460 records = data # type: ignore
1461 if element.name in self.dimension_record_cache:
1462 self.dimension_record_cache.reset()
1463 self._managers.dimensions.insert(
1464 element,
1465 *records,
1466 replace=replace,
1467 skip_existing=skip_existing,
1468 )
1470 def syncDimensionData(
1471 self,
1472 element: DimensionElement | str,
1473 row: Mapping[str, Any] | DimensionRecord,
1474 conform: bool = True,
1475 update: bool = False,
1476 ) -> bool | dict[str, Any]:
1477 """Synchronize the given dimension record with the database, inserting
1478 if it does not already exist and comparing values if it does.
1480 Parameters
1481 ----------
1482 element : `DimensionElement` or `str`
1483 The `DimensionElement` or name thereof that identifies the table
1484 records will be inserted into.
1485 row : `dict` or `DimensionRecord`
1486 The record to insert.
1487 conform : `bool`, optional
1488 If `False` (`True` is default) perform no checking or conversions,
1489 and assume that ``element`` is a `DimensionElement` instance and
1490 ``data`` is a `DimensionRecord` instances of the appropriate
1491 subclass.
1492 update : `bool`, optional
1493 If `True` (`False` is default), update the existing record in the
1494 database if there is a conflict.
1496 Returns
1497 -------
1498 inserted_or_updated : `bool` or `dict`
1499 `True` if a new row was inserted, `False` if no changes were
1500 needed, or a `dict` mapping updated column names to their old
1501 values if an update was performed (only possible if
1502 ``update=True``).
1504 Raises
1505 ------
1506 lsst.daf.butler.registry.ConflictingDefinitionError
1507 Raised if the record exists in the database (according to primary
1508 key lookup) but is inconsistent with the given one.
1509 """
1510 if conform:
1511 if isinstance(element, str):
1512 element = self.dimensions[element]
1513 record = row if isinstance(row, DimensionRecord) else element.RecordClass(**row)
1514 else:
1515 # Ignore typing since caller said to trust them with conform=False.
1516 record = row # type: ignore
1517 if record.definition.name in self.dimension_record_cache:
1518 self.dimension_record_cache.reset()
1519 return self._managers.dimensions.sync(record, update=update)
1521 def queryDatasetTypes(
1522 self,
1523 expression: Any = ...,
1524 *,
1525 missing: list[str] | None = None,
1526 ) -> Iterable[DatasetType]:
1527 """Iterate over the dataset types whose names match an expression.
1529 Parameters
1530 ----------
1531 expression : dataset type expression, optional
1532 An expression that fully or partially identifies the dataset types
1533 to return, such as a `str`, `re.Pattern`, or iterable thereof.
1534 ``...`` can be used to return all dataset types, and is the
1535 default. See :ref:`daf_butler_dataset_type_expressions` for more
1536 information.
1537 missing : `list` of `str`, optional
1538 String dataset type names that were explicitly given (i.e. not
1539 regular expression patterns) but not found will be appended to this
1540 list, if it is provided.
1542 Returns
1543 -------
1544 dataset_types : `~collections.abc.Iterable` [ `DatasetType`]
1545 An `~collections.abc.Iterable` of `DatasetType` instances whose
1546 names match ``expression``.
1548 Raises
1549 ------
1550 lsst.daf.butler.registry.DatasetTypeExpressionError
1551 Raised when ``expression`` is invalid.
1552 """
1553 wildcard = DatasetTypeWildcard.from_expression(expression)
1554 return self._managers.datasets.resolve_wildcard(wildcard, missing=missing)
1556 def queryCollections(
1557 self,
1558 expression: Any = ...,
1559 datasetType: DatasetType | None = None,
1560 collectionTypes: Iterable[CollectionType] | CollectionType = CollectionType.all(),
1561 flattenChains: bool = False,
1562 includeChains: bool | None = None,
1563 ) -> Sequence[str]:
1564 """Iterate over the collections whose names match an expression.
1566 Parameters
1567 ----------
1568 expression : collection expression, optional
1569 An expression that identifies the collections to return, such as
1570 a `str` (for full matches or partial matches via globs),
1571 `re.Pattern` (for partial matches), or iterable thereof. ``...``
1572 can be used to return all collections, and is the default.
1573 See :ref:`daf_butler_collection_expressions` for more information.
1574 datasetType : `DatasetType`, optional
1575 If provided, only yield collections that may contain datasets of
1576 this type. This is a conservative approximation in general; it may
1577 yield collections that do not have any such datasets.
1578 collectionTypes : `~collections.abc.Set` [`CollectionType`] or \
1579 `CollectionType`, optional
1580 If provided, only yield collections of these types.
1581 flattenChains : `bool`, optional
1582 If `True` (`False` is default), recursively yield the child
1583 collections of matching `~CollectionType.CHAINED` collections.
1584 includeChains : `bool`, optional
1585 If `True`, yield records for matching `~CollectionType.CHAINED`
1586 collections. Default is the opposite of ``flattenChains``: include
1587 either CHAINED collections or their children, but not both.
1589 Returns
1590 -------
1591 collections : `~collections.abc.Sequence` [ `str` ]
1592 The names of collections that match ``expression``.
1594 Raises
1595 ------
1596 lsst.daf.butler.registry.CollectionExpressionError
1597 Raised when ``expression`` is invalid.
1599 Notes
1600 -----
1601 The order in which collections are returned is unspecified, except that
1602 the children of a `~CollectionType.CHAINED` collection are guaranteed
1603 to be in the order in which they are searched. When multiple parent
1604 `~CollectionType.CHAINED` collections match the same criteria, the
1605 order in which the two lists appear is unspecified, and the lists of
1606 children may be incomplete if a child has multiple parents.
1607 """
1608 # Right now the datasetTypes argument is completely ignored, but that
1609 # is consistent with its [lack of] guarantees. DM-24939 or a follow-up
1610 # ticket will take care of that.
1611 if datasetType is not None:
1612 warnings.warn(
1613 "The datasetType parameter should no longer be used. It has"
1614 " never had any effect. Will be removed after v28",
1615 FutureWarning,
1616 )
1617 try:
1618 wildcard = CollectionWildcard.from_expression(expression)
1619 except TypeError as exc:
1620 raise CollectionExpressionError(f"Invalid collection expression '{expression}'") from exc
1621 collectionTypes = ensure_iterable(collectionTypes)
1622 return [
1623 record.name
1624 for record in self._managers.collections.resolve_wildcard(
1625 wildcard,
1626 collection_types=frozenset(collectionTypes),
1627 flatten_chains=flattenChains,
1628 include_chains=includeChains,
1629 )
1630 ]
1632 @contextlib.contextmanager
1633 def _query(self) -> Iterator[Query]:
1634 """Context manager returning a `Query` object used for construction
1635 and execution of complex queries.
1636 """
1637 with self._query_driver(self.defaults.collections, self.defaults.dataId) as driver:
1638 yield Query(driver)
1640 @contextlib.contextmanager
1641 def _query_driver(
1642 self,
1643 default_collections: Iterable[str],
1644 default_data_id: DataCoordinate,
1645 ) -> Iterator[DirectQueryDriver]:
1646 """Set up a `QueryDriver` instance for query execution."""
1647 # Query internals do repeated lookups of the same collections, so it
1648 # benefits from the collection record cache.
1649 with self._managers.caching_context.enable_collection_record_cache():
1650 driver = DirectQueryDriver(
1651 self._db,
1652 self.dimensions,
1653 self._managers,
1654 self.dimension_record_cache,
1655 default_collections=default_collections,
1656 default_data_id=default_data_id,
1657 )
1658 with driver:
1659 yield driver
1661 def get_datastore_records(self, ref: DatasetRef) -> DatasetRef:
1662 """Retrieve datastore records for given ref.
1664 Parameters
1665 ----------
1666 ref : `DatasetRef`
1667 Dataset reference for which to retrieve its corresponding datastore
1668 records.
1670 Returns
1671 -------
1672 updated_ref : `DatasetRef`
1673 Dataset reference with filled datastore records.
1675 Notes
1676 -----
1677 If this method is called with the dataset ref that is not known to the
1678 registry then the reference with an empty set of records is returned.
1679 """
1680 datastore_records: dict[str, list[StoredDatastoreItemInfo]] = {}
1681 for opaque, record_class in self._datastore_record_classes.items():
1682 records = self.fetchOpaqueData(opaque, dataset_id=ref.id)
1683 datastore_records[opaque] = [record_class.from_record(record) for record in records]
1684 return ref.replace(datastore_records=datastore_records)
1686 def store_datastore_records(self, refs: Mapping[str, DatasetRef]) -> None:
1687 """Store datastore records for given refs.
1689 Parameters
1690 ----------
1691 refs : `~collections.abc.Mapping` [`str`, `DatasetRef`]
1692 Mapping of a datastore name to dataset reference stored in that
1693 datastore, reference must include datastore records.
1694 """
1695 for datastore_name, ref in refs.items():
1696 # Store ref IDs in the bridge table.
1697 bridge = self._managers.datastores.register(datastore_name)
1698 bridge.insert([ref])
1700 # store records in opaque tables
1701 assert ref._datastore_records is not None, "Dataset ref must have datastore records"
1702 for table_name, records in ref._datastore_records.items():
1703 opaque_table = self._managers.opaque.get(table_name)
1704 assert opaque_table is not None, f"Unexpected opaque table name {table_name}"
1705 opaque_table.insert(*(record.to_record(dataset_id=ref.id) for record in records))
1707 def make_datastore_tables(self, tables: Mapping[str, DatastoreOpaqueTable]) -> None:
1708 """Create opaque tables used by datastores.
1710 Parameters
1711 ----------
1712 tables : `~collections.abc.Mapping`
1713 Maps opaque table name to its definition.
1715 Notes
1716 -----
1717 This method should disappear in the future when opaque table
1718 definitions will be provided during `Registry` construction.
1719 """
1720 datastore_record_classes = {}
1721 for table_name, table_def in tables.items():
1722 datastore_record_classes[table_name] = table_def.record_class
1723 try:
1724 self._managers.opaque.register(table_name, table_def.table_spec)
1725 except ReadOnlyDatabaseError:
1726 # If the database is read only and we just tried and failed to
1727 # create a table, it means someone is trying to create a
1728 # read-only butler client for an empty repo. That should be
1729 # okay, as long as they then try to get any datasets before
1730 # some other client creates the table. Chances are they're
1731 # just validating configuration.
1732 pass
1733 self._datastore_record_classes = datastore_record_classes
1735 def preload_cache(self, *, load_dimension_record_cache: bool) -> None:
1736 """Immediately load caches that are used for common operations.
1738 Parameters
1739 ----------
1740 load_dimension_record_cache : `bool`
1741 If True, preload the dimension record cache. When this cache is
1742 preloaded, subsequent external changes to governor dimension
1743 records will not be visible to this Butler.
1744 """
1745 self._managers.datasets.preload_cache()
1747 if load_dimension_record_cache:
1748 self.dimension_record_cache.preload_cache()
1750 @property
1751 def obsCoreTableManager(self) -> ObsCoreTableManager | None:
1752 """The ObsCore manager instance for this registry
1753 (`~.interfaces.ObsCoreTableManager`
1754 or `None`).
1756 ObsCore manager may not be implemented for all registry backend, or
1757 may not be enabled for many repositories.
1758 """
1759 return self._managers.obscore
1761 storageClasses: StorageClassFactory
1762 """All storage classes known to the registry (`StorageClassFactory`).
1763 """
1765 _defaults: RegistryDefaults
1766 """Default collections used for registry queries (`RegistryDefaults`)."""