Merge pull request #57 from lsst/tickets/DM-39044

DM-39044: Add as_local support for Python resource URIs

doc/changes/DM-39044.feature.rst

@@ -0,0 +1,2 @@
+* Added support for ``as_local`` for Python package resource URIs.
+* Added explicit ``isdir()`` implementation for Python package resources.

python/lsst/resources/packageresource.py

@@ -11,23 +11,31 @@
 
 from __future__ import annotations
 
+__all__ = ("PackageResourcePath",)
+
 import contextlib
 import logging
 import re
 import sys
 
-if sys.version_info >= (3, 11, 0):
-    from importlib import resources
+if sys.version_info < (3, 11, 0):
+    # Mypy will try to use the first import it encounters and ignores
+    # the sys.version_info. This means that the first import has to be
+    # the backwards compatibility import since we are currently using 3.10
+    # for mypy. Once we switch to 3.11 for mypy the order will have to change.
+    import importlib_resources as resources
 else:
-    import importlib_resources as resources  # type: ignore[no-redef]
+    from importlib import resources  # type: ignore[no-redef]
 
-from typing import Iterator
-
-__all__ = ("PackageResourcePath",)
+from collections.abc import Iterator
+from typing import TYPE_CHECKING
 
 from ._resourceHandles._baseResourceHandle import ResourceHandleProtocol
 from ._resourcePath import ResourcePath
 
+if TYPE_CHECKING:
+    import urllib.parse
+
 log = logging.getLogger(__name__)
 
 
@@ -39,17 +47,98 @@ class PackageResourcePath(ResourcePath):
     resource name.
     """
 
+    @classmethod
+    def _fixDirectorySep(
+        cls, parsed: urllib.parse.ParseResult, forceDirectory: bool = False
+    ) -> tuple[urllib.parse.ParseResult, bool]:
+        """Ensure that a path separator is present on directory paths."""
+        parsed, dirLike = super()._fixDirectorySep(parsed, forceDirectory=forceDirectory)
+        if not dirLike:
+            try:
+                # If the resource location does not exist this can
+                # fail immediately. It is possible we are doing path
+                # manipulation and not wanting to read the resource now,
+                # so catch the error and move on.
+                ref = resources.files(parsed.netloc).joinpath(parsed.path.lstrip("/"))
+            except ModuleNotFoundError:
+                pass
+            else:
+                dirLike = ref.is_dir()
+        return parsed, dirLike
+
+    def _get_ref(self) -> resources.abc.Traversable | None:
+        """Obtain the object representing the resource.
+
+        Returns
+        -------
+        path : `resources.abc.Traversable` or `None`
+            The reference to the resource path, or `None` if the module
+            associated with the resources is not accessible. This can happen
+            if Python can't import the Python package defining the resource.
+        """
+        try:
+            ref = resources.files(self.netloc).joinpath(self.relativeToPathRoot)
+        except ModuleNotFoundError:
+            return None
+        return ref
+
+    def isdir(self) -> bool:
+        """Return True if this URI is a directory, else False."""
+        if self.dirLike:  # Always bypass if we guessed the resource is a directory.
+            return True
+        ref = self._get_ref()
+        if ref is None:
+            return False  # Does not seem to exist so assume not a directory.
+        return ref.is_dir()
+
     def exists(self) -> bool:
         """Check that the python resource exists."""
-        ref = resources.files(self.netloc).joinpath(self.relativeToPathRoot)
+        ref = self._get_ref()
+        if ref is None:
+            return False
         return ref.is_file() or ref.is_dir()
 
     def read(self, size: int = -1) -> bytes:
         """Read the contents of the resource."""
-        ref = resources.files(self.netloc).joinpath(self.relativeToPathRoot)
+        ref = self._get_ref()
+        if not ref:
+            raise FileNotFoundError(f"Unable to locate resource {self}.")
         with ref.open("rb") as fh:
             return fh.read(size)
 
+    @contextlib.contextmanager
+    def as_local(self) -> Iterator[ResourcePath]:
+        """Return the location of the Python resource as local file.
+
+        Yields
+        ------
+        local : `ResourcePath`
+            This might be the original resource or a copy on the local file
+            system.
+
+        Notes
+        -----
+        The context manager will automatically delete any local temporary
+        file.
+
+        Examples
+        --------
+        Should be used as a context manager:
+
+        .. code-block:: py
+
+           with uri.as_local() as local:
+               ospath = local.ospath
+        """
+        ref = self._get_ref()
+        if ref is None:
+            raise FileNotFoundError(f"Resource {self} could not be located.")
+        if ref.is_dir():
+            raise IsADirectoryError(f"Directory-like URI {self} cannot be fetched as local.")
+
+        with resources.as_file(ref) as file:
+            yield ResourcePath(file)
+
     @contextlib.contextmanager
     def open(
         self,
@@ -61,21 +150,25 @@ def open(
         # Docstring inherited.
         if "r" not in mode or "+" in mode:
             raise RuntimeError(f"Package resource URI {self} is read-only.")
-        ref = resources.files(self.netloc).joinpath(self.relativeToPathRoot)
+        ref = self._get_ref()
+        if ref is None:
+            raise FileNotFoundError(f"Could not open resource {self}.")
         with ref.open(mode, encoding=encoding) as buffer:
             yield buffer
 
     def walk(
         self, file_filter: str | re.Pattern | None = None
     ) -> Iterator[list | tuple[ResourcePath, list[str], list[str]]]:
         # Docstring inherited.
-        if not self.dirLike:
-            raise ValueError("Can not walk a non-directory URI")
+        if not self.isdir():
+            raise ValueError(f"Can not walk a non-directory URI: {self}")
 
         if isinstance(file_filter, str):
             file_filter = re.compile(file_filter)
 
-        ref = resources.files(self.netloc).joinpath(self.relativeToPathRoot)
+        ref = self._get_ref()
+        if ref is None:
+            raise ValueError(f"Unable to find resource {self}.")
 
         files: list[str] = []
         dirs: list[str] = []

tests/test_resource.py

@@ -43,11 +43,26 @@ def test_read(self):
         content = uri.read().decode()
         self.assertIn("IDLE", content)
 
+        with uri.as_local() as local_uri:
+            self.assertEqual(local_uri.scheme, "file")
+            self.assertTrue(local_uri.exists())
+
         truncated = uri.read(size=9).decode()
         self.assertEqual(truncated, content[:9])
 
+        # Check that directory determination can work directly without the
+        # trailing slash.
+        d = self.root_uri.join("Icons")
+        self.assertTrue(d.isdir())
+        self.assertTrue(d.dirLike)
+
         d = self.root_uri.join("Icons/", forceDirectory=True)
         self.assertTrue(uri.exists(), f"Check directory {d} exists")
+        self.assertTrue(d.isdir())
+
+        with self.assertRaises(IsADirectoryError):
+            with d.as_local() as local_uri:
+                pass
 
         j = d.join("README.txt")
         self.assertEqual(uri, j)
@@ -62,6 +77,18 @@ def test_read(self):
         self.assertFalse(multi[bad])
         self.assertFalse(multi[not_there])
 
+        # Check that the bad URI works as expected.
+        self.assertFalse(bad.exists())
+        self.assertFalse(bad.isdir())
+        with self.assertRaises(FileNotFoundError):
+            bad.read()
+        with self.assertRaises(FileNotFoundError):
+            with bad.as_local():
+                pass
+        with self.assertRaises(FileNotFoundError):
+            with bad.open("r"):
+                pass
+
     def test_open(self):
         uri = self.root_uri.join("Icons/README.txt")
         with uri.open("rb") as buffer:
@@ -113,6 +140,11 @@ def test_walk(self):
         with self.assertRaises(ValueError):
             list(ResourcePath("resource://lsst.resources/http.py").walk())
 
+        bad_dir = ResourcePath(f"{self.scheme}://bad.module/a/dir/")
+        self.assertTrue(bad_dir.isdir())
+        with self.assertRaises(ValueError):
+            list(bad_dir.walk())
+
 
 if __name__ == "__main__":
     unittest.main()