Add butler.collections.register and use it

python/lsst/daf/butler/_butler_collections.py

@@ -270,3 +270,30 @@ def get_info(self, name: str, include_doc: bool = False, include_parents: bool =
             Information on the requested collection.
         """
         raise NotImplementedError()
+
+    @abstractmethod
+    def register(self, name: str, type: CollectionType = CollectionType.RUN, doc: str | None = None) -> bool:
+        """Add a new collection if one with the given name does not exist.
+
+        Parameters
+        ----------
+        name : `str`
+            The name of the collection to create.
+        type : `CollectionType`, optional
+            Enum value indicating the type of collection to create. Default
+            is to create a RUN collection.
+        doc : `str`, optional
+            Documentation string for the collection.
+
+        Returns
+        -------
+        registered : `bool`
+            Boolean indicating whether the collection was already registered
+            or was created by this call.
+
+        Notes
+        -----
+        This method cannot be called within transactions, as it needs to be
+        able to perform its own transaction to be concurrent
+        """
+        raise NotImplementedError()

python/lsst/daf/butler/direct_butler/_direct_butler.py

@@ -1975,7 +1975,7 @@ def transfer_from(
                     if registry := getattr(source_butler, "registry", None):
                         run_doc = registry.getCollectionDocumentation(run)
                     if not dry_run:
-                        registered = self._registry.registerRun(run, doc=run_doc)
+                        registered = self.collections.register(run, doc=run_doc)
                     else:
                         registered = True
                     handled_collections.add(run)

python/lsst/daf/butler/direct_butler/_direct_butler_collections.py

@@ -108,3 +108,6 @@ def get_info(self, name: str, include_doc: bool = False, include_parents: bool =
         if include_parents:
             parents = self._registry.getCollectionParentChains(name)
         return CollectionInfo(name=name, type=record.type, doc=doc, parents=parents, children=children)
+
+    def register(self, name: str, type: CollectionType = CollectionType.RUN, doc: str | None = None) -> bool:
+        return self._registry.registerCollection(name, type, doc)

python/lsst/daf/butler/remote_butler/_remote_butler_collections.py

@@ -95,3 +95,6 @@ def get_info(self, name: str, include_doc: bool = False, include_parents: bool =
         children = info.children or ()
         parents = info.parents or set()
         return CollectionInfo(name=name, type=info.type, doc=doc, parents=parents, children=children)
+
+    def register(self, name: str, type: CollectionType = CollectionType.RUN, doc: str | None = None) -> bool:
+        raise NotImplementedError("Not yet available.")

python/lsst/daf/butler/script/_associate.py

@@ -60,7 +60,7 @@ def associate(
     """
     butler = Butler.from_config(repo, writeable=True, without_datastore=True)
 
-    butler.registry.registerCollection(collection, CollectionType.TAGGED)
+    butler.collections.register(collection, CollectionType.TAGGED)
 
     results = QueryDatasets(
         butler=butler,

python/lsst/daf/butler/script/certifyCalibrations.py

@@ -87,5 +87,5 @@ def certifyCalibrations(
             raise RuntimeError(
                 f"No inputs found for dataset {dataset_type_name} in {input_collection}. {explanation}"
             )
-    butler.registry.registerCollection(output_collection, type=CollectionType.CALIBRATION)
+    butler.collections.register(output_collection, type=CollectionType.CALIBRATION)
     butler.registry.certify(output_collection, refs, timespan)

python/lsst/daf/butler/script/collectionChain.py

@@ -85,7 +85,7 @@ def collectionChain(
         if mode in ("redefine", "extend", "prepend"):
             if not doc:
                 doc = None
-            butler.registry.registerCollection(parent, CollectionType.CHAINED, doc)
+            butler.collections.register(parent, CollectionType.CHAINED, doc)
         else:
             raise RuntimeError(
                 f"Mode '{mode}' requires that the collection exists "

python/lsst/daf/butler/tests/utils.py

@@ -281,7 +281,7 @@ def _do_init(self, butler: Butler, butlerConfigFile: str) -> None:
 
         # New datasets will be added to run and tag, but we will only look in
         # tag when looking up datasets.
-        self.butler.registry.registerCollection(self._DEFAULT_TAG, CollectionType.TAGGED)
+        self.butler.collections.register(self._DEFAULT_TAG, CollectionType.TAGGED)
 
         # Create and register a DatasetType
         self.datasetType = addDatasetType(
@@ -351,7 +351,7 @@ def addDataset(
             A reference to the added dataset.
         """
         if run:
-            self.butler.registry.registerCollection(run, type=CollectionType.RUN)
+            self.butler.collections.register(run)
         else:
             run = self._DEFAULT_RUN
         metric = self._makeExampleMetrics()