Add butler.collections.x_remove

python/lsst/daf/butler/_butler_collections.py

@@ -297,3 +297,36 @@ def register(self, name: str, type: CollectionType = CollectionType.RUN, doc: st
         able to perform its own transaction to be concurrent
         """
         raise NotImplementedError()
+
+    @abstractmethod
+    def x_remove(self, name: str) -> None:
+        """Remove the given collection from the registry.
+
+        Parameters
+        ----------
+        name : `str`
+            The name of the collection to remove.
+
+        Raises
+        ------
+        lsst.daf.butler.registry.MissingCollectionError
+            Raised if no collection with the given name exists.
+        sqlalchemy.exc.IntegrityError
+            Raised if the database rows associated with the collection are
+            still referenced by some other table, such as a dataset in a
+            datastore (for `~CollectionType.RUN` collections only) or a
+            `~CollectionType.CHAINED` collection of which this collection is
+            a child.
+
+        Notes
+        -----
+        If this is a `~CollectionType.RUN` collection, all datasets and quanta
+        in it will removed from the `Registry` database.  This requires that
+        those datasets be removed (or at least trashed) from any datastores
+        that hold them first.
+
+        A collection may not be deleted as long as it is referenced by a
+        `~CollectionType.CHAINED` collection; the ``CHAINED`` collection must
+        be deleted or redefined first.
+        """
+        raise NotImplementedError()

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

@@ -111,3 +111,6 @@ def get_info(self, name: str, include_doc: bool = False, include_parents: bool =
 
     def register(self, name: str, type: CollectionType = CollectionType.RUN, doc: str | None = None) -> bool:
         return self._registry.registerCollection(name, type, doc)
+
+    def x_remove(self, name: str) -> None:
+        self._registry.removeCollection(name)

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

@@ -98,3 +98,6 @@ def get_info(self, name: str, include_doc: bool = False, include_parents: bool =
 
     def register(self, name: str, type: CollectionType = CollectionType.RUN, doc: str | None = None) -> bool:
         raise NotImplementedError("Not yet available.")
+
+    def x_remove(self, name: str) -> None:
+        raise NotImplementedError("Not yet available.")

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

@@ -126,7 +126,7 @@ def _doRemove(collections: Table) -> None:
         """Perform the prune collection step."""
         butler = Butler.from_config(repo, writeable=True, without_datastore=True)
         for name in collections["Collection"]:
-            butler.registry.removeCollection(name)
+            butler.collections.x_remove(name)
 
     result = RemoveCollectionResult(
         onConfirmation=partial(_doRemove, collectionInfo.nonRunCollections),