Add butler.collections.register and register_run

python/lsst/daf/butler/_butler_collections.py

@@ -270,3 +270,52 @@ 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, 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`
+            Enum value indicating the type of collection to create.
+        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()
+
+    def register_run(self, name: str, doc: str | None = None) -> bool:
+        """Add a new run if one with the given name does not exist.
+
+        Parameters
+        ----------
+        name : `str`
+            The name of the run to create.
+        doc : `str`, optional
+            Documentation string for the collection.
+
+        Returns
+        -------
+        registered : `bool`
+            Boolean indicating whether a new run was registered. `False`
+            if it already existed.
+
+        Notes
+        -----
+        This method cannot be called within transactions, as it needs to be
+        able to perform its own transaction to be concurrent.
+        """
+        return self.register(name, CollectionType.RUN, doc)

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, 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, doc: str | None = None) -> bool:
+        raise NotImplementedError("Not yet available.")