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