@contextlib.contextmanager
-def _tempfile(reader, suffix='',
- # gh-93353: Keep a reference to call os.remove() in late Python
- # finalization.
- *, _os_remove=os.remove):
+def _tempfile(
+ reader,
+ suffix='',
+ # gh-93353: Keep a reference to call os.remove() in late Python
+ # finalization.
+ *,
+ _os_remove=os.remove,
+):
# Not using tempfile.NamedTemporaryFile as it leads to deeper 'try'
# blocks due to the need to close the temporary file to work on Windows
# properly.
pass
+def _temp_file(path):
+ return _tempfile(path.read_bytes, suffix=path.name)
+
+
+def _is_present_dir(path: Traversable) -> bool:
+ """
+ Some Traversables implement ``is_dir()`` to raise an
+ exception (i.e. ``FileNotFoundError``) when the
+ directory doesn't exist. This function wraps that call
+ to always return a boolean and only return True
+ if there's a dir and it exists.
+ """
+ with contextlib.suppress(FileNotFoundError):
+ return path.is_dir()
+ return False
+
+
@functools.singledispatch
def as_file(path):
"""
Given a Traversable object, return that object as a
path on the local file system in a context manager.
"""
- return _tempfile(path.read_bytes, suffix=path.name)
+ return _temp_dir(path) if _is_present_dir(path) else _temp_file(path)
@as_file.register(pathlib.Path)
Degenerate behavior for pathlib.Path objects.
"""
yield path
+
+
+@contextlib.contextmanager
+def _temp_path(dir: tempfile.TemporaryDirectory):
+ """
+ Wrap tempfile.TemporyDirectory to return a pathlib object.
+ """
+ with dir as result:
+ yield pathlib.Path(result)
+
+
+@contextlib.contextmanager
+def _temp_dir(path):
+ """
+ Given a traversable dir, recursively replicate the whole tree
+ to the file system in a context manager.
+ """
+ assert path.is_dir()
+ with _temp_path(tempfile.TemporaryDirectory()) as temp_dir:
+ yield _write_contents(temp_dir, path)
+
+
+def _write_contents(target, source):
+ child = target.joinpath(source.name)
+ if source.is_dir():
+ child.mkdir()
+ for item in source.iterdir():
+ _write_contents(child, item)
+ else:
+ child.open('wb').write(source.read_bytes())
+ return child
import abc
import io
+import itertools
import os
+import pathlib
from typing import Any, BinaryIO, Iterable, Iterator, NoReturn, Text, Optional
from typing import runtime_checkable, Protocol
from typing import Union
raise FileNotFoundError
+class TraversalError(Exception):
+ pass
+
+
@runtime_checkable
class Traversable(Protocol):
"""
Return True if self is a file
"""
- @abc.abstractmethod
def joinpath(self, *descendants: StrPath) -> "Traversable":
"""
Return Traversable resolved with any descendants applied.
and each may contain multiple levels separated by
``posixpath.sep`` (``/``).
"""
+ if not descendants:
+ return self
+ names = itertools.chain.from_iterable(
+ path.parts for path in map(pathlib.PurePosixPath, descendants)
+ )
+ target = next(names)
+ matches = (
+ traversable for traversable in self.iterdir() if traversable.name == target
+ )
+ try:
+ match = next(matches)
+ except StopIteration:
+ raise TraversalError(
+ "Target not found during traversal.", target, list(names)
+ )
+ return match.joinpath(*names)
def __truediv__(self, child: StrPath) -> "Traversable":
"""
def is_file(self):
return False
- def joinpath(self, child):
- # first try to find child in current paths
- for file in self.iterdir():
- if file.name == child:
- return file
- # if it does not exist, construct it with the first path
- return self._paths[0] / child
-
- __truediv__ = joinpath
+ def joinpath(self, *descendants):
+ try:
+ return super().joinpath(*descendants)
+ except abc.TraversalError:
+ # One of the paths did not resolve (a directory does not exist).
+ # Just return something that will not exist.
+ return self._paths[0].joinpath(*descendants)
def open(self, *args, **kwargs):
raise FileNotFoundError(f'{self} is not a file')
def open(self, *args, **kwargs):
raise IsADirectoryError()
- @staticmethod
- def _flatten(compound_names):
- for name in compound_names:
- yield from name.split('/')
-
- def joinpath(self, *descendants):
- if not descendants:
- return self
- names = self._flatten(descendants)
- target = next(names)
- return next(
- traversable for traversable in self.iterdir() if traversable.name == target
- ).joinpath(*names)
-
class TraversableReader(TraversableResources, SimpleReader):
"""
str(path.joinpath('imaginary'))[len(prefix) + 1 :],
os.path.join('namespacedata01', 'imaginary'),
)
+ self.assertEqual(path.joinpath(), path)
+
+ def test_join_path_compound(self):
+ path = MultiplexedPath(self.folder)
+ assert not path.joinpath('imaginary/foo.py').exists()
def test_repr(self):
self.assertEqual(
{'__init__.py', 'binary.file'},
)
+ def test_as_file_directory(self):
+ with resources.as_file(resources.files('ziptestdata')) as data:
+ assert data.name == 'ziptestdata'
+ assert data.is_dir()
+ assert data.joinpath('subdirectory').is_dir()
+ assert len(list(data.iterdir()))
+ assert not data.parent.exists()
+
class ResourceFromZipsTest02(util.ZipSetupBase, unittest.TestCase):
ZIP_MODULE = zipdata02 # type: ignore
--- /dev/null
+Apply changes from importlib_resources 5.8 and 5.9: ``Traversable.joinpath``
+provides a concrete implementation. ``as_file`` now supports directories of
+resources.
projects / thirdparty / Python / cpython.git / commitdiff
