diff options
| author | shadchin <[email protected]> | 2024-02-12 07:53:52 +0300 |
|---|---|---|
| committer | Daniil Cherednik <[email protected]> | 2024-02-14 14:26:16 +0000 |
| commit | 31f2a419764a8ba77c2a970cfc80056c6cd06756 (patch) | |
| tree | c1995d239eba8571cefc640f6648e1d5dd4ce9e2 /contrib/tools/python3/src/Lib/importlib/resources | |
| parent | fe2ef02b38d9c85d80060963b265a1df9f38c3bb (diff) | |
Update Python from 3.11.8 to 3.12.2
Diffstat (limited to 'contrib/tools/python3/src/Lib/importlib/resources')
7 files changed, 252 insertions, 127 deletions
diff --git a/contrib/tools/python3/src/Lib/importlib/resources/_adapters.py b/contrib/tools/python3/src/Lib/importlib/resources/_adapters.py index ea363d86a56..50688fbb666 100644 --- a/contrib/tools/python3/src/Lib/importlib/resources/_adapters.py +++ b/contrib/tools/python3/src/Lib/importlib/resources/_adapters.py @@ -34,9 +34,7 @@ def _io_wrapper(file, mode='r', *args, **kwargs): return TextIOWrapper(file, *args, **kwargs) elif mode == 'rb': return file - raise ValueError( - "Invalid mode value '{}', only 'r' and 'rb' are supported".format(mode) - ) + raise ValueError(f"Invalid mode value '{mode}', only 'r' and 'rb' are supported") class CompatibilityFiles: diff --git a/contrib/tools/python3/src/Lib/importlib/resources/_common.py b/contrib/tools/python3/src/Lib/importlib/resources/_common.py index ca1fa8ab2fe..a3902535342 100644 --- a/contrib/tools/python3/src/Lib/importlib/resources/_common.py +++ b/contrib/tools/python3/src/Lib/importlib/resources/_common.py @@ -5,25 +5,58 @@ import functools import contextlib import types import importlib +import inspect +import warnings +import itertools -from typing import Union, Optional +from typing import Union, Optional, cast from .abc import ResourceReader, Traversable from ._adapters import wrap_spec Package = Union[types.ModuleType, str] +Anchor = Package -def files(package): - # type: (Package) -> Traversable +def package_to_anchor(func): """ - Get a Traversable resource from a package + Replace 'package' parameter as 'anchor' and warn about the change. + + Other errors should fall through. + + >>> files('a', 'b') + Traceback (most recent call last): + TypeError: files() takes from 0 to 1 positional arguments but 2 were given """ - return from_package(get_package(package)) + undefined = object() + + @functools.wraps(func) + def wrapper(anchor=undefined, package=undefined): + if package is not undefined: + if anchor is not undefined: + return func(anchor, package) + warnings.warn( + "First parameter to files is renamed to 'anchor'", + DeprecationWarning, + stacklevel=2, + ) + return func(package) + elif anchor is undefined: + return func() + return func(anchor) + return wrapper -def get_resource_reader(package): - # type: (types.ModuleType) -> Optional[ResourceReader] + +@package_to_anchor +def files(anchor: Optional[Anchor] = None) -> Traversable: + """ + Get a Traversable resource for an anchor. + """ + return from_package(resolve(anchor)) + + +def get_resource_reader(package: types.ModuleType) -> Optional[ResourceReader]: """ Return the package's loader if it's a ResourceReader. """ @@ -39,24 +72,39 @@ def get_resource_reader(package): return reader(spec.name) # type: ignore -def resolve(cand): - # type: (Package) -> types.ModuleType - return cand if isinstance(cand, types.ModuleType) else importlib.import_module(cand) +def resolve(cand: Optional[Anchor]) -> types.ModuleType: + return cast(types.ModuleType, cand) + + +def _(cand: str) -> types.ModuleType: + return importlib.import_module(cand) + +def _(cand: None) -> types.ModuleType: + return resolve(_infer_caller().f_globals['__name__']) -def get_package(package): - # type: (Package) -> types.ModuleType - """Take a package name or module object and return the module. - Raise an exception if the resolved module is not a package. +def _infer_caller(): """ - resolved = resolve(package) - if wrap_spec(resolved).submodule_search_locations is None: - raise TypeError(f'{package!r} is not a package') - return resolved + Walk the stack and find the frame of the first caller not in this module. + """ + + def is_this_file(frame_info): + return frame_info.filename == __file__ + def is_wrapper(frame_info): + return frame_info.function == 'wrapper' -def from_package(package): + not_this_file = itertools.filterfalse(is_this_file, inspect.stack()) + # also exclude 'wrapper' due to singledispatch in the call stack + callers = itertools.filterfalse(is_wrapper, not_this_file) + return next(callers).frame + + +def from_package(package: types.ModuleType): """ Return a Traversable object for the given package. @@ -67,10 +115,14 @@ def from_package(package): @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. @@ -89,13 +141,30 @@ def _tempfile(reader, suffix='', 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) @@ -105,3 +174,34 @@ def _(path): Degenerate behavior for pathlib.Path objects. """ yield path + + +def _temp_path(dir: tempfile.TemporaryDirectory): + """ + Wrap tempfile.TemporyDirectory to return a pathlib object. + """ + with dir as result: + yield pathlib.Path(result) + + +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.write_bytes(source.read_bytes()) + return child diff --git a/contrib/tools/python3/src/Lib/importlib/resources/_itertools.py b/contrib/tools/python3/src/Lib/importlib/resources/_itertools.py index cce05582ffc..7b775ef5ae8 100644 --- a/contrib/tools/python3/src/Lib/importlib/resources/_itertools.py +++ b/contrib/tools/python3/src/Lib/importlib/resources/_itertools.py @@ -1,35 +1,38 @@ -from itertools import filterfalse +# from more_itertools 9.0 +def only(iterable, default=None, too_long=None): + """If *iterable* has only one item, return it. + If it has zero items, return *default*. + If it has more than one item, raise the exception given by *too_long*, + which is ``ValueError`` by default. + >>> only([], default='missing') + 'missing' + >>> only([1]) + 1 + >>> only([1, 2]) # doctest: +IGNORE_EXCEPTION_DETAIL + Traceback (most recent call last): + ... + ValueError: Expected exactly one item in iterable, but got 1, 2, + and perhaps more.' + >>> only([1, 2], too_long=TypeError) # doctest: +IGNORE_EXCEPTION_DETAIL + Traceback (most recent call last): + ... + TypeError + Note that :func:`only` attempts to advance *iterable* twice to ensure there + is only one item. See :func:`spy` or :func:`peekable` to check + iterable contents less destructively. + """ + it = iter(iterable) + first_value = next(it, default) -from typing import ( - Callable, - Iterable, - Iterator, - Optional, - Set, - TypeVar, - Union, -) - -# Type and type variable definitions -_T = TypeVar('_T') -_U = TypeVar('_U') - - -def unique_everseen( - iterable: Iterable[_T], key: Optional[Callable[[_T], _U]] = None -) -> Iterator[_T]: - "List unique elements, preserving order. Remember all elements ever seen." - # unique_everseen('AAAABBBCCDAABBB') --> A B C D - # unique_everseen('ABBCcAD', str.lower) --> A B C D - seen: Set[Union[_T, _U]] = set() - seen_add = seen.add - if key is None: - for element in filterfalse(seen.__contains__, iterable): - seen_add(element) - yield element + try: + second_value = next(it) + except StopIteration: + pass else: - for element in iterable: - k = key(element) - if k not in seen: - seen_add(k) - yield element + msg = ( + 'Expected exactly one item in iterable, but got {!r}, {!r}, ' + 'and perhaps more.'.format(first_value, second_value) + ) + raise too_long or ValueError(msg) + + return first_value diff --git a/contrib/tools/python3/src/Lib/importlib/resources/_legacy.py b/contrib/tools/python3/src/Lib/importlib/resources/_legacy.py index 1d5d3f1fbb1..b1ea8105dad 100644 --- a/contrib/tools/python3/src/Lib/importlib/resources/_legacy.py +++ b/contrib/tools/python3/src/Lib/importlib/resources/_legacy.py @@ -27,8 +27,7 @@ def deprecated(func): return wrapper -def normalize_path(path): - # type: (Any) -> str +def normalize_path(path: Any) -> str: """Normalize a path by ensuring it is a string. If the resulting string contains path separators, an exception is raised. diff --git a/contrib/tools/python3/src/Lib/importlib/resources/abc.py b/contrib/tools/python3/src/Lib/importlib/resources/abc.py index 0b7bfdc4158..6750a7aaf14 100644 --- a/contrib/tools/python3/src/Lib/importlib/resources/abc.py +++ b/contrib/tools/python3/src/Lib/importlib/resources/abc.py @@ -1,6 +1,8 @@ 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 @@ -53,6 +55,10 @@ class ResourceReader(metaclass=abc.ABCMeta): raise FileNotFoundError +class TraversalError(Exception): + pass + + @runtime_checkable class Traversable(Protocol): """ @@ -95,7 +101,6 @@ 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. @@ -104,6 +109,22 @@ class Traversable(Protocol): 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": """ @@ -121,7 +142,8 @@ class Traversable(Protocol): accepted by io.TextIOWrapper. """ - @abc.abstractproperty + @property + @abc.abstractmethod def name(self) -> str: """ The base name of this object without any parent references. diff --git a/contrib/tools/python3/src/Lib/importlib/resources/readers.py b/contrib/tools/python3/src/Lib/importlib/resources/readers.py index b470a2062b2..c3cdf769cbe 100644 --- a/contrib/tools/python3/src/Lib/importlib/resources/readers.py +++ b/contrib/tools/python3/src/Lib/importlib/resources/readers.py @@ -1,11 +1,12 @@ import collections -import operator +import itertools import pathlib +import operator import zipfile from . import abc -from ._itertools import unique_everseen +from ._itertools import only def remove_duplicates(items): @@ -41,8 +42,10 @@ class ZipReader(abc.TraversableResources): raise FileNotFoundError(exc.args[0]) def is_resource(self, path): - # workaround for `zipfile.Path.is_file` returning true - # for non-existent paths. + """ + Workaround for `zipfile.Path.is_file` returning true + for non-existent paths. + """ target = self.files().joinpath(path) return target.is_file() and target.exists() @@ -67,8 +70,10 @@ class MultiplexedPath(abc.Traversable): raise NotADirectoryError('MultiplexedPath only supports directories') def iterdir(self): - files = (file for path in self._paths for file in path.iterdir()) - return unique_everseen(files, key=operator.attrgetter('name')) + children = (child for path in self._paths for child in path.iterdir()) + by_name = operator.attrgetter('name') + groups = itertools.groupby(sorted(children, key=by_name), key=by_name) + return map(self._follow, (locs for name, locs in groups)) def read_bytes(self): raise FileNotFoundError(f'{self} is not a file') @@ -82,15 +87,32 @@ class MultiplexedPath(abc.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 + 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) + + @classmethod + def _follow(cls, children): + """ + Construct a MultiplexedPath if needed. - __truediv__ = joinpath + If children contains a sole element, return it. + Otherwise, return a MultiplexedPath of the items. + Unless one of the items is not a Directory, then return the first. + """ + subdirs, one_dir, one_file = itertools.tee(children, 3) + + try: + return only(one_dir) + except ValueError: + try: + return cls(*subdirs) + except NotADirectoryError: + return next(one_file) def open(self, *args, **kwargs): raise FileNotFoundError(f'{self} is not a file') diff --git a/contrib/tools/python3/src/Lib/importlib/resources/simple.py b/contrib/tools/python3/src/Lib/importlib/resources/simple.py index d0fbf237762..7770c922c84 100644 --- a/contrib/tools/python3/src/Lib/importlib/resources/simple.py +++ b/contrib/tools/python3/src/Lib/importlib/resources/simple.py @@ -16,31 +16,28 @@ class SimpleReader(abc.ABC): provider. """ - @abc.abstractproperty - def package(self): - # type: () -> str + @property + @abc.abstractmethod + def package(self) -> str: """ The name of the package for which this reader loads resources. """ @abc.abstractmethod - def children(self): - # type: () -> List['SimpleReader'] + def children(self) -> List['SimpleReader']: """ Obtain an iterable of SimpleReader for available child containers (e.g. directories). """ @abc.abstractmethod - def resources(self): - # type: () -> List[str] + def resources(self) -> List[str]: """ Obtain available named resources for this virtual package. """ @abc.abstractmethod - def open_binary(self, resource): - # type: (str) -> BinaryIO + def open_binary(self, resource: str) -> BinaryIO: """ Obtain a File-like for a named resource. """ @@ -50,39 +47,12 @@ class SimpleReader(abc.ABC): return self.package.split('.')[-1] -class ResourceHandle(Traversable): - """ - Handle to a named resource in a ResourceReader. - """ - - def __init__(self, parent, name): - # type: (ResourceContainer, str) -> None - self.parent = parent - self.name = name # type: ignore - - def is_file(self): - return True - - def is_dir(self): - return False - - def open(self, mode='r', *args, **kwargs): - stream = self.parent.reader.open_binary(self.name) - if 'b' not in mode: - stream = io.TextIOWrapper(*args, **kwargs) - return stream - - def joinpath(self, name): - raise RuntimeError("Cannot traverse into a resource") - - class ResourceContainer(Traversable): """ Traversable container for a package's resources via its reader. """ - def __init__(self, reader): - # type: (SimpleReader) -> None + def __init__(self, reader: SimpleReader): self.reader = reader def is_dir(self): @@ -99,19 +69,30 @@ class ResourceContainer(Traversable): 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 ResourceHandle(Traversable): + """ + Handle to a named resource in a ResourceReader. + """ + + def __init__(self, parent: ResourceContainer, name: str): + self.parent = parent + self.name = name # type: ignore + + def is_file(self): + return True + + def is_dir(self): + return False + + def open(self, mode='r', *args, **kwargs): + stream = self.parent.reader.open_binary(self.name) + if 'b' not in mode: + stream = io.TextIOWrapper(*args, **kwargs) + return stream + + def joinpath(self, name): + raise RuntimeError("Cannot traverse into a resource") class TraversableReader(TraversableResources, SimpleReader): |
