Coverage for python/lsst/daf/butler/registry/sql_registry.py: 22%
408 statements
« prev ^ index » next coverage.py v7.11.0, created at 2025-11-20 01:50 -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 refs = self._managers.datasets.get_dataset_refs([id])
1028 if len(refs) == 0:
1029 return None
1030 else:
1031 return refs[0]
1033 def _fetch_run_dataset_ids(self, run: str) -> list[DatasetId]:
1034 """Return the IDs of all datasets in the given ``RUN``
1035 collection.
1037 Parameters
1038 ----------
1039 run : `str`
1040 Name of the collection.
1042 Returns
1043 -------
1044 dataset_ids : `list` [`uuid.UUID`]
1045 List of dataset IDs.
1047 Notes
1048 -----
1049 This is a middleware-internal interface.
1050 """
1051 run_record = self._managers.collections.find(run)
1052 if not isinstance(run_record, RunRecord):
1053 raise CollectionTypeError(f"{run!r} is not a RUN collection.")
1054 return self._managers.datasets.fetch_run_dataset_ids(run_record)
1056 @transactional
1057 def removeDatasets(self, refs: Iterable[DatasetRef]) -> None:
1058 """Remove datasets from the Registry.
1060 The datasets will be removed unconditionally from all collections.
1061 `Datastore` records will *not* be deleted; the caller is responsible
1062 for ensuring that the dataset has already been removed from all
1063 Datastores.
1065 Parameters
1066 ----------
1067 refs : `~collections.abc.Iterable` [`DatasetRef`]
1068 References to the datasets to be removed. Should be considered
1069 invalidated upon return.
1071 Raises
1072 ------
1073 lsst.daf.butler.registry.OrphanedRecordError
1074 Raised if any dataset is still present in any `Datastore`.
1075 """
1076 try:
1077 self._managers.datasets.delete(refs)
1078 except sqlalchemy.exc.IntegrityError as err:
1079 raise OrphanedRecordError(
1080 "One or more datasets is still present in one or more Datastores."
1081 ) from err
1083 @transactional
1084 def associate(self, collection: str, refs: Iterable[DatasetRef]) -> None:
1085 """Add existing datasets to a `~CollectionType.TAGGED` collection.
1087 If a DatasetRef with the same exact ID is already in a collection
1088 nothing is changed. If a `DatasetRef` with the same `DatasetType` and
1089 data ID but with different ID exists in the collection,
1090 `~lsst.daf.butler.registry.ConflictingDefinitionError` is raised.
1092 Parameters
1093 ----------
1094 collection : `str`
1095 Indicates the collection the datasets should be associated with.
1096 refs : `~collections.abc.Iterable` [ `DatasetRef` ]
1097 An iterable of resolved `DatasetRef` instances that already exist
1098 in this `Registry`.
1100 Raises
1101 ------
1102 lsst.daf.butler.registry.ConflictingDefinitionError
1103 If a Dataset with the given `DatasetRef` already exists in the
1104 given collection.
1105 lsst.daf.butler.registry.MissingCollectionError
1106 Raised if ``collection`` does not exist in the registry.
1107 lsst.daf.butler.registry.CollectionTypeError
1108 Raise adding new datasets to the given ``collection`` is not
1109 allowed.
1110 """
1111 progress = Progress("lsst.daf.butler.Registry.associate", level=logging.DEBUG)
1112 collectionRecord = self._managers.collections.find(collection)
1113 for datasetType, refsForType in progress.iter_item_chunks(
1114 DatasetRef.iter_by_type(refs), desc="Associating datasets by type"
1115 ):
1116 try:
1117 self._managers.datasets.associate(datasetType, collectionRecord, refsForType)
1118 if self._managers.obscore:
1119 # If a TAGGED collection is being monitored by ObsCore
1120 # manager then we may need to save the dataset.
1121 self._managers.obscore.associate(refsForType, collectionRecord)
1122 except sqlalchemy.exc.IntegrityError as err:
1123 raise ConflictingDefinitionError(
1124 f"Constraint violation while associating dataset of type {datasetType.name} with "
1125 f"collection {collection}. This probably means that one or more datasets with the same "
1126 "dataset type and data ID already exist in the collection, but it may also indicate "
1127 "that the datasets do not exist."
1128 ) from err
1130 @transactional
1131 def disassociate(self, collection: str, refs: Iterable[DatasetRef]) -> None:
1132 """Remove existing datasets from a `~CollectionType.TAGGED` collection.
1134 ``collection`` and ``ref`` combinations that are not currently
1135 associated are silently ignored.
1137 Parameters
1138 ----------
1139 collection : `str`
1140 The collection the datasets should no longer be associated with.
1141 refs : `~collections.abc.Iterable` [ `DatasetRef` ]
1142 An iterable of resolved `DatasetRef` instances that already exist
1143 in this `Registry`.
1145 Raises
1146 ------
1147 lsst.daf.butler.AmbiguousDatasetError
1148 Raised if any of the given dataset references is unresolved.
1149 lsst.daf.butler.registry.MissingCollectionError
1150 Raised if ``collection`` does not exist in the registry.
1151 lsst.daf.butler.registry.CollectionTypeError
1152 Raise adding new datasets to the given ``collection`` is not
1153 allowed.
1154 """
1155 progress = Progress("lsst.daf.butler.Registry.disassociate", level=logging.DEBUG)
1156 collectionRecord = self._managers.collections.find(collection)
1157 for datasetType, refsForType in progress.iter_item_chunks(
1158 DatasetRef.iter_by_type(refs), desc="Disassociating datasets by type"
1159 ):
1160 self._managers.datasets.disassociate(datasetType, collectionRecord, refsForType)
1161 if self._managers.obscore:
1162 self._managers.obscore.disassociate(refsForType, collectionRecord)
1164 @transactional
1165 def certify(self, collection: str, refs: Iterable[DatasetRef], timespan: Timespan) -> None:
1166 """Associate one or more datasets with a calibration collection and a
1167 validity range within it.
1169 Parameters
1170 ----------
1171 collection : `str`
1172 The name of an already-registered `~CollectionType.CALIBRATION`
1173 collection.
1174 refs : `~collections.abc.Iterable` [ `DatasetRef` ]
1175 Datasets to be associated.
1176 timespan : `Timespan`
1177 The validity range for these datasets within the collection.
1179 Raises
1180 ------
1181 lsst.daf.butler.AmbiguousDatasetError
1182 Raised if any of the given `DatasetRef` instances is unresolved.
1183 lsst.daf.butler.registry.ConflictingDefinitionError
1184 Raised if the collection already contains a different dataset with
1185 the same `DatasetType` and data ID and an overlapping validity
1186 range.
1187 DatasetTypeError
1188 Raised if ``ref.datasetType.isCalibration() is False`` for any ref.
1189 CollectionTypeError
1190 Raised if
1191 ``collection.type is not CollectionType.CALIBRATION``.
1192 """
1193 progress = Progress("lsst.daf.butler.Registry.certify", level=logging.DEBUG)
1194 with self._managers.caching_context.enable_collection_record_cache():
1195 collectionRecord = self._managers.collections.find(collection)
1196 for datasetType, refsForType in progress.iter_item_chunks(
1197 DatasetRef.iter_by_type(refs), desc="Certifying datasets by type"
1198 ):
1199 self._managers.datasets.certify(
1200 datasetType, collectionRecord, refsForType, timespan, self._query
1201 )
1203 @transactional
1204 def decertify(
1205 self,
1206 collection: str,
1207 datasetType: str | DatasetType,
1208 timespan: Timespan,
1209 *,
1210 dataIds: Iterable[DataId] | None = None,
1211 ) -> None:
1212 """Remove or adjust datasets to clear a validity range within a
1213 calibration collection.
1215 Parameters
1216 ----------
1217 collection : `str`
1218 The name of an already-registered `~CollectionType.CALIBRATION`
1219 collection.
1220 datasetType : `str` or `DatasetType`
1221 Name or `DatasetType` instance for the datasets to be decertified.
1222 timespan : `Timespan`, optional
1223 The validity range to remove datasets from within the collection.
1224 Datasets that overlap this range but are not contained by it will
1225 have their validity ranges adjusted to not overlap it, which may
1226 split a single dataset validity range into two.
1227 dataIds : iterable [`dict` or `DataCoordinate`], optional
1228 Data IDs that should be decertified within the given validity range
1229 If `None`, all data IDs for ``self.datasetType`` will be
1230 decertified.
1232 Raises
1233 ------
1234 DatasetTypeError
1235 Raised if ``datasetType.isCalibration() is False``.
1236 CollectionTypeError
1237 Raised if
1238 ``collection.type is not CollectionType.CALIBRATION``.
1239 """
1240 collectionRecord = self._managers.collections.find(collection)
1241 if isinstance(datasetType, str):
1242 datasetType = self.getDatasetType(datasetType)
1243 standardizedDataIds = None
1244 if dataIds is not None:
1245 standardizedDataIds = [
1246 DataCoordinate.standardize(d, dimensions=datasetType.dimensions) for d in dataIds
1247 ]
1248 self._managers.datasets.decertify(
1249 datasetType, collectionRecord, timespan, data_ids=standardizedDataIds, query_func=self._query
1250 )
1252 def getDatastoreBridgeManager(self) -> DatastoreRegistryBridgeManager:
1253 """Return an object that allows a new `Datastore` instance to
1254 communicate with this `Registry`.
1256 Returns
1257 -------
1258 manager : `~.interfaces.DatastoreRegistryBridgeManager`
1259 Object that mediates communication between this `Registry` and its
1260 associated datastores.
1261 """
1262 return self._managers.datastores
1264 def getDatasetLocations(self, ref: DatasetRef) -> Iterable[str]:
1265 """Retrieve datastore locations for a given dataset.
1267 Parameters
1268 ----------
1269 ref : `DatasetRef`
1270 A reference to the dataset for which to retrieve storage
1271 information.
1273 Returns
1274 -------
1275 datastores : `~collections.abc.Iterable` [ `str` ]
1276 All the matching datastores holding this dataset.
1278 Raises
1279 ------
1280 lsst.daf.butler.AmbiguousDatasetError
1281 Raised if ``ref.id`` is `None`.
1282 """
1283 return self._managers.datastores.findDatastores(ref)
1285 def expandDataId(
1286 self,
1287 dataId: DataId | None = None,
1288 *,
1289 dimensions: Iterable[str] | DimensionGroup | None = None,
1290 records: Mapping[str, DimensionRecord | None] | None = None,
1291 withDefaults: bool = True,
1292 **kwargs: Any,
1293 ) -> DataCoordinate:
1294 """Expand a dimension-based data ID to include additional information.
1296 Parameters
1297 ----------
1298 dataId : `DataCoordinate` or `dict`, optional
1299 Data ID to be expanded; augmented and overridden by ``kwargs``.
1300 dimensions : `~collections.abc.Iterable` [ `str` ], \
1301 `DimensionGroup`, optional
1302 The dimensions to be identified by the new `DataCoordinate`.
1303 If not provided, will be inferred from the keys of ``dataId`` and
1304 ``**kwargs``, and ``universe`` must be provided unless ``dataId``
1305 is already a `DataCoordinate`.
1306 records : `~collections.abc.Mapping` [`str`, `DimensionRecord`], \
1307 optional
1308 Dimension record data to use before querying the database for that
1309 data, keyed by element name.
1310 withDefaults : `bool`, optional
1311 Utilize ``self.defaults.dataId`` to fill in missing governor
1312 dimension key-value pairs. Defaults to `True` (i.e. defaults are
1313 used).
1314 **kwargs
1315 Additional keywords are treated like additional key-value pairs for
1316 ``dataId``, extending and overriding.
1318 Returns
1319 -------
1320 expanded : `DataCoordinate`
1321 A data ID that includes full metadata for all of the dimensions it
1322 identifies, i.e. guarantees that ``expanded.hasRecords()`` and
1323 ``expanded.hasFull()`` both return `True`.
1325 Raises
1326 ------
1327 lsst.daf.butler.registry.DataIdError
1328 Raised when ``dataId`` or keyword arguments specify unknown
1329 dimensions or values, or when a resulting data ID contains
1330 contradictory key-value pairs, according to dimension
1331 relationships.
1333 Notes
1334 -----
1335 This method cannot be relied upon to reject invalid data ID values
1336 for dimensions that do actually not have any record columns. For
1337 efficiency reasons the records for these dimensions (which have only
1338 dimension key values that are given by the caller) may be constructed
1339 directly rather than obtained from the registry database.
1340 """
1341 if not withDefaults:
1342 defaults = None
1343 else:
1344 defaults = self.defaults.dataId
1345 standardized = DataCoordinate.standardize(
1346 dataId,
1347 dimensions=dimensions,
1348 universe=self.dimensions,
1349 defaults=defaults,
1350 **kwargs,
1351 )
1352 if standardized.hasRecords():
1353 return standardized
1354 if records is None:
1355 records = {}
1356 else:
1357 records = dict(records)
1358 if isinstance(dataId, DataCoordinate) and dataId.hasRecords():
1359 for element_name in dataId.dimensions.elements:
1360 records[element_name] = dataId.records[element_name]
1361 keys: dict[str, str | int] = dict(standardized.mapping)
1362 for element_name in standardized.dimensions.lookup_order:
1363 element = self.dimensions[element_name]
1364 record = records.get(element_name, ...) # Use ... to mean not found; None might mean NULL
1365 if record is ...:
1366 if element_name in self.dimensions.dimensions.names and keys.get(element_name) is None:
1367 raise DimensionNameError(f"No value or null value for dimension {element_name}.")
1368 else:
1369 record = self._managers.dimensions.fetch_one(
1370 element_name,
1371 DataCoordinate.standardize(keys, dimensions=element.minimal_group),
1372 self.dimension_record_cache,
1373 )
1374 records[element_name] = record
1375 if record is not None:
1376 for d in element.implied:
1377 value = getattr(record, d.name)
1378 if keys.setdefault(d.name, value) != value:
1379 raise InconsistentDataIdError(
1380 f"Data ID {standardized} has {d.name}={keys[d.name]!r}, "
1381 f"but {element_name} implies {d.name}={value!r}."
1382 )
1383 else:
1384 if element_name in standardized.dimensions.names:
1385 raise DataIdValueError(
1386 f"Could not fetch record for dimension {element.name} via keys {keys}."
1387 )
1388 if element.defines_relationships:
1389 raise InconsistentDataIdError(
1390 f"Could not fetch record for element {element_name} via keys {keys}, "
1391 "but it is marked as defining relationships; this means one or more dimensions are "
1392 "have inconsistent values.",
1393 )
1394 return DataCoordinate.standardize(keys, dimensions=standardized.dimensions).expanded(records=records)
1396 def expand_data_ids(self, data_ids: Iterable[DataCoordinate]) -> list[DataCoordinate]:
1397 output = list(data_ids)
1399 grouped_by_dimensions: defaultdict[DimensionGroup, list[int]] = defaultdict(list)
1400 for i, data_id in enumerate(data_ids):
1401 if not data_id.hasRecords():
1402 grouped_by_dimensions[data_id.dimensions].append(i)
1404 if not grouped_by_dimensions:
1405 # All given DataCoordinate values are already expanded.
1406 return output
1408 attacher = DimensionDataAttacher(
1409 cache=self.dimension_record_cache,
1410 dimensions=DimensionGroup.union(*grouped_by_dimensions.keys(), universe=self.dimensions),
1411 )
1412 with self._query() as query:
1413 for dimensions, indexes in grouped_by_dimensions.items():
1414 expanded = attacher.attach(dimensions, (output[index] for index in indexes), query)
1415 for index, data_id in zip(indexes, expanded):
1416 output[index] = data_id
1418 return output
1420 def expand_refs(self, dataset_refs: list[DatasetRef]) -> list[DatasetRef]:
1421 expanded_ids = self.expand_data_ids([ref.dataId for ref in dataset_refs])
1422 return [ref.expanded(data_id) for ref, data_id in zip(dataset_refs, expanded_ids)]
1424 def insertDimensionData(
1425 self,
1426 element: DimensionElement | str,
1427 *data: Mapping[str, Any] | DimensionRecord,
1428 conform: bool = True,
1429 replace: bool = False,
1430 skip_existing: bool = False,
1431 ) -> None:
1432 """Insert one or more dimension records into the database.
1434 Parameters
1435 ----------
1436 element : `DimensionElement` or `str`
1437 The `DimensionElement` or name thereof that identifies the table
1438 records will be inserted into.
1439 *data : `dict` or `DimensionRecord`
1440 One or more records to insert.
1441 conform : `bool`, optional
1442 If `False` (`True` is default) perform no checking or conversions,
1443 and assume that ``element`` is a `DimensionElement` instance and
1444 ``data`` is a one or more `DimensionRecord` instances of the
1445 appropriate subclass.
1446 replace : `bool`, optional
1447 If `True` (`False` is default), replace existing records in the
1448 database if there is a conflict.
1449 skip_existing : `bool`, optional
1450 If `True` (`False` is default), skip insertion if a record with
1451 the same primary key values already exists. Unlike
1452 `syncDimensionData`, this will not detect when the given record
1453 differs from what is in the database, and should not be used when
1454 this is a concern.
1455 """
1456 if isinstance(element, str):
1457 element = self.dimensions[element]
1458 if conform:
1459 records = [
1460 row if isinstance(row, DimensionRecord) else element.RecordClass(**row) for row in data
1461 ]
1462 else:
1463 # Ignore typing since caller said to trust them with conform=False.
1464 records = data # type: ignore
1465 if element.name in self.dimension_record_cache:
1466 self.dimension_record_cache.reset()
1467 self._managers.dimensions.insert(
1468 element,
1469 *records,
1470 replace=replace,
1471 skip_existing=skip_existing,
1472 )
1474 def syncDimensionData(
1475 self,
1476 element: DimensionElement | str,
1477 row: Mapping[str, Any] | DimensionRecord,
1478 conform: bool = True,
1479 update: bool = False,
1480 ) -> bool | dict[str, Any]:
1481 """Synchronize the given dimension record with the database, inserting
1482 if it does not already exist and comparing values if it does.
1484 Parameters
1485 ----------
1486 element : `DimensionElement` or `str`
1487 The `DimensionElement` or name thereof that identifies the table
1488 records will be inserted into.
1489 row : `dict` or `DimensionRecord`
1490 The record to insert.
1491 conform : `bool`, optional
1492 If `False` (`True` is default) perform no checking or conversions,
1493 and assume that ``element`` is a `DimensionElement` instance and
1494 ``data`` is a `DimensionRecord` instances of the appropriate
1495 subclass.
1496 update : `bool`, optional
1497 If `True` (`False` is default), update the existing record in the
1498 database if there is a conflict.
1500 Returns
1501 -------
1502 inserted_or_updated : `bool` or `dict`
1503 `True` if a new row was inserted, `False` if no changes were
1504 needed, or a `dict` mapping updated column names to their old
1505 values if an update was performed (only possible if
1506 ``update=True``).
1508 Raises
1509 ------
1510 lsst.daf.butler.registry.ConflictingDefinitionError
1511 Raised if the record exists in the database (according to primary
1512 key lookup) but is inconsistent with the given one.
1513 """
1514 if conform:
1515 if isinstance(element, str):
1516 element = self.dimensions[element]
1517 record = row if isinstance(row, DimensionRecord) else element.RecordClass(**row)
1518 else:
1519 # Ignore typing since caller said to trust them with conform=False.
1520 record = row # type: ignore
1521 if record.definition.name in self.dimension_record_cache:
1522 self.dimension_record_cache.reset()
1523 return self._managers.dimensions.sync(record, update=update)
1525 def queryDatasetTypes(
1526 self,
1527 expression: Any = ...,
1528 *,
1529 missing: list[str] | None = None,
1530 ) -> Iterable[DatasetType]:
1531 """Iterate over the dataset types whose names match an expression.
1533 Parameters
1534 ----------
1535 expression : dataset type expression, optional
1536 An expression that fully or partially identifies the dataset types
1537 to return, such as a `str`, `re.Pattern`, or iterable thereof.
1538 ``...`` can be used to return all dataset types, and is the
1539 default. See :ref:`daf_butler_dataset_type_expressions` for more
1540 information.
1541 missing : `list` of `str`, optional
1542 String dataset type names that were explicitly given (i.e. not
1543 regular expression patterns) but not found will be appended to this
1544 list, if it is provided.
1546 Returns
1547 -------
1548 dataset_types : `~collections.abc.Iterable` [ `DatasetType`]
1549 An `~collections.abc.Iterable` of `DatasetType` instances whose
1550 names match ``expression``.
1552 Raises
1553 ------
1554 lsst.daf.butler.registry.DatasetTypeExpressionError
1555 Raised when ``expression`` is invalid.
1556 """
1557 wildcard = DatasetTypeWildcard.from_expression(expression)
1558 return self._managers.datasets.resolve_wildcard(wildcard, missing=missing)
1560 def queryCollections(
1561 self,
1562 expression: Any = ...,
1563 datasetType: DatasetType | None = None,
1564 collectionTypes: Iterable[CollectionType] | CollectionType = CollectionType.all(),
1565 flattenChains: bool = False,
1566 includeChains: bool | None = None,
1567 ) -> Sequence[str]:
1568 """Iterate over the collections whose names match an expression.
1570 Parameters
1571 ----------
1572 expression : collection expression, optional
1573 An expression that identifies the collections to return, such as
1574 a `str` (for full matches or partial matches via globs),
1575 `re.Pattern` (for partial matches), or iterable thereof. ``...``
1576 can be used to return all collections, and is the default.
1577 See :ref:`daf_butler_collection_expressions` for more information.
1578 datasetType : `DatasetType`, optional
1579 If provided, only yield collections that may contain datasets of
1580 this type. This is a conservative approximation in general; it may
1581 yield collections that do not have any such datasets.
1582 collectionTypes : `~collections.abc.Set` [`CollectionType`] or \
1583 `CollectionType`, optional
1584 If provided, only yield collections of these types.
1585 flattenChains : `bool`, optional
1586 If `True` (`False` is default), recursively yield the child
1587 collections of matching `~CollectionType.CHAINED` collections.
1588 includeChains : `bool`, optional
1589 If `True`, yield records for matching `~CollectionType.CHAINED`
1590 collections. Default is the opposite of ``flattenChains``: include
1591 either CHAINED collections or their children, but not both.
1593 Returns
1594 -------
1595 collections : `~collections.abc.Sequence` [ `str` ]
1596 The names of collections that match ``expression``.
1598 Raises
1599 ------
1600 lsst.daf.butler.registry.CollectionExpressionError
1601 Raised when ``expression`` is invalid.
1603 Notes
1604 -----
1605 The order in which collections are returned is unspecified, except that
1606 the children of a `~CollectionType.CHAINED` collection are guaranteed
1607 to be in the order in which they are searched. When multiple parent
1608 `~CollectionType.CHAINED` collections match the same criteria, the
1609 order in which the two lists appear is unspecified, and the lists of
1610 children may be incomplete if a child has multiple parents.
1611 """
1612 # Right now the datasetTypes argument is completely ignored, but that
1613 # is consistent with its [lack of] guarantees. DM-24939 or a follow-up
1614 # ticket will take care of that.
1615 if datasetType is not None:
1616 warnings.warn(
1617 "The datasetType parameter should no longer be used. It has"
1618 " never had any effect. Will be removed after v28",
1619 FutureWarning,
1620 )
1621 try:
1622 wildcard = CollectionWildcard.from_expression(expression)
1623 except TypeError as exc:
1624 raise CollectionExpressionError(f"Invalid collection expression '{expression}'") from exc
1625 collectionTypes = ensure_iterable(collectionTypes)
1626 return [
1627 record.name
1628 for record in self._managers.collections.resolve_wildcard(
1629 wildcard,
1630 collection_types=frozenset(collectionTypes),
1631 flatten_chains=flattenChains,
1632 include_chains=includeChains,
1633 )
1634 ]
1636 @contextlib.contextmanager
1637 def _query(self) -> Iterator[Query]:
1638 """Context manager returning a `Query` object used for construction
1639 and execution of complex queries.
1640 """
1641 with self._query_driver(self.defaults.collections, self.defaults.dataId) as driver:
1642 yield Query(driver)
1644 @contextlib.contextmanager
1645 def _query_driver(
1646 self,
1647 default_collections: Iterable[str],
1648 default_data_id: DataCoordinate,
1649 ) -> Iterator[DirectQueryDriver]:
1650 """Set up a `QueryDriver` instance for query execution."""
1651 # Query internals do repeated lookups of the same collections, so it
1652 # benefits from the collection record cache.
1653 with self._managers.caching_context.enable_collection_record_cache():
1654 driver = DirectQueryDriver(
1655 self._db,
1656 self.dimensions,
1657 self._managers,
1658 self.dimension_record_cache,
1659 default_collections=default_collections,
1660 default_data_id=default_data_id,
1661 )
1662 with driver:
1663 yield driver
1665 def get_datastore_records(self, ref: DatasetRef) -> DatasetRef:
1666 """Retrieve datastore records for given ref.
1668 Parameters
1669 ----------
1670 ref : `DatasetRef`
1671 Dataset reference for which to retrieve its corresponding datastore
1672 records.
1674 Returns
1675 -------
1676 updated_ref : `DatasetRef`
1677 Dataset reference with filled datastore records.
1679 Notes
1680 -----
1681 If this method is called with the dataset ref that is not known to the
1682 registry then the reference with an empty set of records is returned.
1683 """
1684 datastore_records: dict[str, list[StoredDatastoreItemInfo]] = {}
1685 for opaque, record_class in self._datastore_record_classes.items():
1686 records = self.fetchOpaqueData(opaque, dataset_id=ref.id)
1687 datastore_records[opaque] = [record_class.from_record(record) for record in records]
1688 return ref.replace(datastore_records=datastore_records)
1690 def store_datastore_records(self, refs: Mapping[str, DatasetRef]) -> None:
1691 """Store datastore records for given refs.
1693 Parameters
1694 ----------
1695 refs : `~collections.abc.Mapping` [`str`, `DatasetRef`]
1696 Mapping of a datastore name to dataset reference stored in that
1697 datastore, reference must include datastore records.
1698 """
1699 for datastore_name, ref in refs.items():
1700 # Store ref IDs in the bridge table.
1701 bridge = self._managers.datastores.register(datastore_name)
1702 bridge.insert([ref])
1704 # store records in opaque tables
1705 assert ref._datastore_records is not None, "Dataset ref must have datastore records"
1706 for table_name, records in ref._datastore_records.items():
1707 opaque_table = self._managers.opaque.get(table_name)
1708 assert opaque_table is not None, f"Unexpected opaque table name {table_name}"
1709 opaque_table.insert(*(record.to_record(dataset_id=ref.id) for record in records))
1711 def make_datastore_tables(self, tables: Mapping[str, DatastoreOpaqueTable]) -> None:
1712 """Create opaque tables used by datastores.
1714 Parameters
1715 ----------
1716 tables : `~collections.abc.Mapping`
1717 Maps opaque table name to its definition.
1719 Notes
1720 -----
1721 This method should disappear in the future when opaque table
1722 definitions will be provided during `Registry` construction.
1723 """
1724 datastore_record_classes = {}
1725 for table_name, table_def in tables.items():
1726 datastore_record_classes[table_name] = table_def.record_class
1727 try:
1728 self._managers.opaque.register(table_name, table_def.table_spec)
1729 except ReadOnlyDatabaseError:
1730 # If the database is read only and we just tried and failed to
1731 # create a table, it means someone is trying to create a
1732 # read-only butler client for an empty repo. That should be
1733 # okay, as long as they then try to get any datasets before
1734 # some other client creates the table. Chances are they're
1735 # just validating configuration.
1736 pass
1737 self._datastore_record_classes = datastore_record_classes
1739 def preload_cache(self, *, load_dimension_record_cache: bool) -> None:
1740 """Immediately load caches that are used for common operations.
1742 Parameters
1743 ----------
1744 load_dimension_record_cache : `bool`
1745 If True, preload the dimension record cache. When this cache is
1746 preloaded, subsequent external changes to governor dimension
1747 records will not be visible to this Butler.
1748 """
1749 self._managers.datasets.preload_cache()
1751 if load_dimension_record_cache:
1752 self.dimension_record_cache.preload_cache()
1754 @property
1755 def obsCoreTableManager(self) -> ObsCoreTableManager | None:
1756 """The ObsCore manager instance for this registry
1757 (`~.interfaces.ObsCoreTableManager`
1758 or `None`).
1760 ObsCore manager may not be implemented for all registry backend, or
1761 may not be enabled for many repositories.
1762 """
1763 return self._managers.obscore
1765 storageClasses: StorageClassFactory
1766 """All storage classes known to the registry (`StorageClassFactory`).
1767 """
1769 _defaults: RegistryDefaults
1770 """Default collections used for registry queries (`RegistryDefaults`)."""