implement file io methods

Author: rickeylev

python/runfiles/runfiles.py

@@ -29,7 +29,7 @@
 import posixpath
 import sys
 from collections import defaultdict
-from typing import Dict, List, Optional, Tuple, Union
+from typing import Dict, Generator, Iterable, List, Optional, Tuple, Union
 
 
 class _RepositoryMapping:
@@ -143,25 +143,25 @@ def is_empty(self) -> bool:
         )
 
 
-class Path(pathlib.PurePath):
+class Path(pathlib.Path):
     """A pathlib-like path object for runfiles.
 
-    This class extends `pathlib.PurePath` and resolves paths
+    This class extends `pathlib.Path` and resolves paths
     using the associated `Runfiles` instance when converted to a string.
     """
 
-    # For Python < 3.12 compatibility when subclassing PurePath directly
-    _flavour = getattr(type(pathlib.PurePath()), "_flavour", None)
+    # For Python < 3.12 compatibility when subclassing Path directly
+    _flavour = getattr(type(pathlib.Path()), "_flavour", None)
 
     def __new__(
         cls,
         *args: Union[str, os.PathLike],
         runfiles: Optional["Runfiles"] = None,
         source_repo: Optional[str] = None,
-    ) -> "Path":
+    ) -> "Self":
         """Private constructor. Use Runfiles.root() to create instances."""
         obj = super().__new__(cls, *args)
-        # Type checkers might complain about adding attributes to PurePath,
+        # Type checkers might complain about adding attributes to Path,
         # but this is standard for pathlib subclasses.
         obj._runfiles = runfiles  # type: ignore
         obj._source_repo = source_repo  # type: ignore
@@ -173,90 +173,205 @@ def __init__(
         runfiles: Optional["Runfiles"] = None,
         source_repo: Optional[str] = None,
     ) -> None:
-        pass
+        # In Python 3.12+, pathlib was refactored and Path.__init__ now accepts
+        # *args. Prior to 3.12, Path did not define __init__, so
+        # super().__init__(*args) would fall through to object.__init__, which
+        # raises a TypeError because it takes no arguments.
+        if sys.version_info >= (3, 12):
+            super().__init__(*args)
+        else:
+            super().__init__()
+
+    # We override resolve() and absolute() to ensure that in Python < 3.12,
+    # where pathlib internally uses object.__new__ instead of our custom
+    # __new__ or with_segments(), the runfiles state is preserved. We delegate
+    # to self._as_path() because super().resolve() creates intermediate objects
+    # that would otherwise crash during internal stat() calls.
+    # override
+    def resolve(self, strict: bool = False) -> "Self":
+        return type(self)(
+            self._as_path().resolve(strict=strict),
+            runfiles=self._runfiles,
+            source_repo=self._source_repo,
+        )
+
+    # override
+    def absolute(self) -> "Self":
+        return type(self)(
+            self._as_path().absolute(),
+            runfiles=self._runfiles,
+            source_repo=self._source_repo,
+        )
 
-    def with_segments(self, *pathsegments: Union[str, os.PathLike]) -> "Path":
+    # override
+    def with_segments(self, *pathsegments: Union[str, os.PathLike]) -> "Self":
         """Used by Python 3.12+ pathlib to create new path objects."""
         return type(self)(
             *pathsegments,
-            runfiles=self._runfiles,  # type: ignore
-            source_repo=self._source_repo,  # type: ignore
+            runfiles=self._runfiles,
+            source_repo=self._source_repo,
         )
 
     # For Python < 3.12
-    @classmethod
-    def _from_parts(cls, args: Tuple[str, ...]) -> "Path":
-        obj = super()._from_parts(args)  # type: ignore
-        # These will be set by the calling instance later, or we can't set them here
-        # properly without context. Usually pathlib calls this from an instance
-        # method like _make_child, which we also might need to override.
-        return obj
-
-    def _make_child(self, args: Tuple[str, ...]) -> "Path":
+    # override
+    def _make_child(self, args: Tuple[str, ...]) -> "Self":
         obj = super()._make_child(args)  # type: ignore
         obj._runfiles = self._runfiles  # type: ignore
         obj._source_repo = self._source_repo  # type: ignore
         return obj
 
-    @classmethod
-    def _from_parsed_parts(cls, drv: str, root: str, parts: List[str]) -> "Path":
-        obj = super()._from_parsed_parts(drv, root, parts)  # type: ignore
-        return obj
-
-    def _make_child_relpath(self, part: str) -> "Path":
-        obj = super()._make_child_relpath(part)  # type: ignore
-        obj._runfiles = self._runfiles  # type: ignore
-        obj._source_repo = self._source_repo  # type: ignore
-        return obj
-
+    # override
     @property
-    def parents(self) -> Tuple["Path", ...]:
+    def parents(self) -> Tuple["Self", ...]:
         return tuple(
             type(self)(
                 p,
-                runfiles=getattr(self, "_runfiles", None),
-                source_repo=getattr(self, "_source_repo", None),
+                runfiles=self._runfiles,
+                source_repo=self._source_repo,
             )
             for p in super().parents
         )
 
+    # override
     @property
-    def parent(self) -> "Path":
+    def parent(self) -> "Self":
         return type(self)(
             super().parent,
-            runfiles=getattr(self, "_runfiles", None),
-            source_repo=getattr(self, "_source_repo", None),
+            runfiles=self._runfiles,
+            source_repo=self._source_repo,
         )
 
-    def with_name(self, name: str) -> "Path":
+    @property
+    def runfile_path(self) -> str:
+        """Returns the runfiles-root relative path."""
+        path_posix = super().__str__().replace("\\", "/")
+        if path_posix == ".":
+            return ""
+        return path_posix
+
+    # override
+    def with_name(self, name: str) -> "Self":
         return type(self)(
             super().with_name(name),
-            runfiles=getattr(self, "_runfiles", None),
-            source_repo=getattr(self, "_source_repo", None),
+            runfiles=self._runfiles,
+            source_repo=self._source_repo,
         )
 
-    def with_suffix(self, suffix: str) -> "Path":
+    # override
+    def with_suffix(self, suffix: str) -> "Self":
         return type(self)(
             super().with_suffix(suffix),
-            runfiles=getattr(self, "_runfiles", None),
-            source_repo=getattr(self, "_source_repo", None),
+            runfiles=self._runfiles,
+            source_repo=self._source_repo,
         )
 
+    def _as_path(self) -> pathlib.Path:
+        return pathlib.Path(str(self))
+
+    # override
+    def stat(self, *, follow_symlinks: bool = True) -> os.stat_result:
+        return self._as_path().stat(follow_symlinks=follow_symlinks)
+
+    # override
+    def lstat(self) -> os.stat_result:
+        return self._as_path().lstat()
+
+    # override
+    def exists(self) -> bool:
+        return self._as_path().exists()
+
+    # override
+    def is_dir(self) -> bool:
+        return self._as_path().is_dir()
+
+    # override
+    def is_file(self) -> bool:
+        return self._as_path().is_file()
+
+    # override
+    def is_symlink(self) -> bool:
+        return self._as_path().is_symlink()
+
+    # override
+    def is_block_device(self) -> bool:
+        return self._as_path().is_block_device()
+
+    # override
+    def is_char_device(self) -> bool:
+        return self._as_path().is_char_device()
+
+    # override
+    def is_fifo(self) -> bool:
+        return self._as_path().is_fifo()
+
+    # override
+    def is_socket(self) -> bool:
+        return self._as_path().is_socket()
+
+    # override
+    def open(
+        self,
+        mode: str = "r",
+        buffering: int = -1,
+        encoding: Optional[str] = None,
+        errors: Optional[str] = None,
+        newline: Optional[str] = None,
+    ):
+        return self._as_path().open(
+            mode=mode,
+            buffering=buffering,
+            encoding=encoding,
+            errors=errors,
+            newline=newline,
+        )
+
+    # override
+    def read_bytes(self) -> bytes:
+        return self._as_path().read_bytes()
+
+    # override
+    def read_text(
+        self, encoding: Optional[str] = None, errors: Optional[str] = None
+    ) -> str:
+        return self._as_path().read_text(encoding=encoding, errors=errors)
+
+    # override
+    def iterdir(self) -> Generator["Self", None, None]:
+        resolved = self._as_path()
+        for p in resolved.iterdir():
+            yield self / p.name
+
+    # override
+    def glob(self, pattern: str) -> Generator["Self", None, None]:
+        resolved = self._as_path()
+        for p in resolved.glob(pattern):
+            yield self / p.relative_to(resolved)
+
+    # override
+    def rglob(self, pattern: str) -> Generator["Self", None, None]:
+        resolved = self._as_path()
+        for p in resolved.rglob(pattern):
+            yield self / p.relative_to(resolved)
+
     def __repr__(self) -> str:
-        return 'runfiles.Path({!r})'.format(super().__str__())
+        return 'runfiles.Path({!r})'.format(self.runfile_path)
 
     def __str__(self) -> str:
         path_posix = super().__str__().replace("\\", "/")
         if not path_posix or path_posix == ".":
             # pylint: disable=protected-access
             return self._runfiles._python_runfiles_root  # type: ignore
         resolved = self._runfiles.Rlocation(path_posix, source_repo=self._source_repo)  # type: ignore
-        return resolved if resolved is not None else super().__str__()
+        if resolved is not None:
+            return resolved
+
+        # pylint: disable=protected-access
+        return posixpath.join(self._runfiles._python_runfiles_root, path_posix)  # type: ignore
 
     def __fspath__(self) -> str:
         return str(self)
 
-    def runfiles_root(self) -> "Path":
+    def runfiles_root(self) -> "Self":
         """Returns a Path object representing the runfiles root."""
         return self._runfiles.root(source_repo=self._source_repo)  # type: ignore