Use FormatterV2 for configs

python/lsst/pipe/base/formatters/pexConfig.py

@@ -29,55 +29,26 @@
 
 __all__ = ("PexConfigFormatter",)
 
-import os.path
-from typing import Any
+from typing import Any, BinaryIO
 
-from lsst.daf.butler.formatters.file import FileFormatter
+from lsst.daf.butler import FormatterV2
 from lsst.pex.config import Config
+from lsst.resources import ResourceHandleProtocol, ResourcePath
 
 
-class PexConfigFormatter(FileFormatter):
+class PexConfigFormatter(FormatterV2):
     """Formatter implementation for reading and writing
     `lsst.pex.config.Config` instances.
     """
 
-    extension = ".py"
-
-    def _readFile(self, path: str, pytype: type[Any] | None = None) -> Any:
-        """Read a pex.config.Config instance from the given file.
-
-        Parameters
-        ----------
-        path : `str`
-            Path to use to open the file.
-        pytype : `type`, optional
-            Class to use to read the config file.
-
-        Returns
-        -------
-        data : `lsst.pex.config.Config`
-            Instance of class ``pytype`` read from config file. `None`
-            if the file could not be opened.
-        """
-        if not os.path.exists(path):
-            return None
+    default_extension = ".py"
+    can_read_from_stream = True
 
+    def read_from_stream(
+        self, stream: BinaryIO | ResourceHandleProtocol, component: str | None = None, expected_size: int = -1
+    ) -> Any:
         # Automatically determine the Config class from the serialized form
-        with open(path, "r") as fd:
-            config_py = fd.read()
-        return Config._fromPython(config_py)
-
-    def _writeFile(self, inMemoryDataset: Any) -> None:
-        """Write the in memory dataset to file on disk.
-
-        Parameters
-        ----------
-        inMemoryDataset : `object`
-            Object to serialize.
+        return Config._fromPython(stream.read().decode())
 
-        Raises
-        ------
-        Exception
-            The file could not be written.
-        """
-        inMemoryDataset.save(self.fileDescriptor.location.path)
+    def write_local_file(self, in_memory_dataset: Any, uri: ResourcePath) -> None:
+        in_memory_dataset.save(uri.path)