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