summaryrefslogtreecommitdiffstats
path: root/contrib/tools/python3/src/Lib/importlib/resources
diff options
context:
space:
mode:
authorAlexSm <[email protected]>2024-03-05 10:40:59 +0100
committerGitHub <[email protected]>2024-03-05 12:40:59 +0300
commit1ac13c847b5358faba44dbb638a828e24369467b (patch)
tree07672b4dd3604ad3dee540a02c6494cb7d10dc3d /contrib/tools/python3/src/Lib/importlib/resources
parentffcca3e7f7958ddc6487b91d3df8c01054bd0638 (diff)
Library import 16 (#2433)
Co-authored-by: robot-piglet <[email protected]> Co-authored-by: deshevoy <[email protected]> Co-authored-by: robot-contrib <[email protected]> Co-authored-by: thegeorg <[email protected]> Co-authored-by: robot-ya-builder <[email protected]> Co-authored-by: svidyuk <[email protected]> Co-authored-by: shadchin <[email protected]> Co-authored-by: robot-ratatosk <[email protected]> Co-authored-by: innokentii <[email protected]> Co-authored-by: arkady-e1ppa <[email protected]> Co-authored-by: snermolaev <[email protected]> Co-authored-by: dimdim11 <[email protected]> Co-authored-by: kickbutt <[email protected]> Co-authored-by: abdullinsaid <[email protected]> Co-authored-by: korsunandrei <[email protected]> Co-authored-by: petrk <[email protected]> Co-authored-by: miroslav2 <[email protected]> Co-authored-by: serjflint <[email protected]> Co-authored-by: akhropov <[email protected]> Co-authored-by: prettyboy <[email protected]> Co-authored-by: ilikepugs <[email protected]> Co-authored-by: hiddenpath <[email protected]> Co-authored-by: mikhnenko <[email protected]> Co-authored-by: spreis <[email protected]> Co-authored-by: andreyshspb <[email protected]> Co-authored-by: dimaandreev <[email protected]> Co-authored-by: rashid <[email protected]> Co-authored-by: robot-ydb-importer <[email protected]> Co-authored-by: r-vetrov <[email protected]> Co-authored-by: ypodlesov <[email protected]> Co-authored-by: zaverden <[email protected]> Co-authored-by: vpozdyayev <[email protected]> Co-authored-by: robot-cozmo <[email protected]> Co-authored-by: v-korovin <[email protected]> Co-authored-by: arikon <[email protected]> Co-authored-by: khoden <[email protected]> Co-authored-by: psydmm <[email protected]> Co-authored-by: robot-javacom <[email protected]> Co-authored-by: dtorilov <[email protected]> Co-authored-by: sennikovmv <[email protected]> Co-authored-by: hcpp <[email protected]>
Diffstat (limited to 'contrib/tools/python3/src/Lib/importlib/resources')
-rw-r--r--contrib/tools/python3/src/Lib/importlib/resources/__init__.py36
-rw-r--r--contrib/tools/python3/src/Lib/importlib/resources/_adapters.py168
-rw-r--r--contrib/tools/python3/src/Lib/importlib/resources/_common.py207
-rw-r--r--contrib/tools/python3/src/Lib/importlib/resources/_itertools.py38
-rw-r--r--contrib/tools/python3/src/Lib/importlib/resources/_legacy.py120
-rw-r--r--contrib/tools/python3/src/Lib/importlib/resources/abc.py173
-rw-r--r--contrib/tools/python3/src/Lib/importlib/resources/readers.py144
-rw-r--r--contrib/tools/python3/src/Lib/importlib/resources/simple.py106
8 files changed, 0 insertions, 992 deletions
diff --git a/contrib/tools/python3/src/Lib/importlib/resources/__init__.py b/contrib/tools/python3/src/Lib/importlib/resources/__init__.py
deleted file mode 100644
index 34e3a9950cc..00000000000
--- a/contrib/tools/python3/src/Lib/importlib/resources/__init__.py
+++ /dev/null
@@ -1,36 +0,0 @@
-"""Read resources contained within a package."""
-
-from ._common import (
- as_file,
- files,
- Package,
-)
-
-from ._legacy import (
- contents,
- open_binary,
- read_binary,
- open_text,
- read_text,
- is_resource,
- path,
- Resource,
-)
-
-from .abc import ResourceReader
-
-
-__all__ = [
- 'Package',
- 'Resource',
- 'ResourceReader',
- 'as_file',
- 'contents',
- 'files',
- 'is_resource',
- 'open_binary',
- 'open_text',
- 'path',
- 'read_binary',
- 'read_text',
-]
diff --git a/contrib/tools/python3/src/Lib/importlib/resources/_adapters.py b/contrib/tools/python3/src/Lib/importlib/resources/_adapters.py
deleted file mode 100644
index 50688fbb666..00000000000
--- a/contrib/tools/python3/src/Lib/importlib/resources/_adapters.py
+++ /dev/null
@@ -1,168 +0,0 @@
-from contextlib import suppress
-from io import TextIOWrapper
-
-from . import abc
-
-
-class SpecLoaderAdapter:
- """
- Adapt a package spec to adapt the underlying loader.
- """
-
- def __init__(self, spec, adapter=lambda spec: spec.loader):
- self.spec = spec
- self.loader = adapter(spec)
-
- def __getattr__(self, name):
- return getattr(self.spec, name)
-
-
-class TraversableResourcesLoader:
- """
- Adapt a loader to provide TraversableResources.
- """
-
- def __init__(self, spec):
- self.spec = spec
-
- def get_resource_reader(self, name):
- return CompatibilityFiles(self.spec)._native()
-
-
-def _io_wrapper(file, mode='r', *args, **kwargs):
- if mode == 'r':
- return TextIOWrapper(file, *args, **kwargs)
- elif mode == 'rb':
- return file
- raise ValueError(f"Invalid mode value '{mode}', only 'r' and 'rb' are supported")
-
-
-class CompatibilityFiles:
- """
- Adapter for an existing or non-existent resource reader
- to provide a compatibility .files().
- """
-
- class SpecPath(abc.Traversable):
- """
- Path tied to a module spec.
- Can be read and exposes the resource reader children.
- """
-
- def __init__(self, spec, reader):
- self._spec = spec
- self._reader = reader
-
- def iterdir(self):
- if not self._reader:
- return iter(())
- return iter(
- CompatibilityFiles.ChildPath(self._reader, path)
- for path in self._reader.contents()
- )
-
- def is_file(self):
- return False
-
- is_dir = is_file
-
- def joinpath(self, other):
- if not self._reader:
- return CompatibilityFiles.OrphanPath(other)
- return CompatibilityFiles.ChildPath(self._reader, other)
-
- @property
- def name(self):
- return self._spec.name
-
- def open(self, mode='r', *args, **kwargs):
- return _io_wrapper(self._reader.open_resource(None), mode, *args, **kwargs)
-
- class ChildPath(abc.Traversable):
- """
- Path tied to a resource reader child.
- Can be read but doesn't expose any meaningful children.
- """
-
- def __init__(self, reader, name):
- self._reader = reader
- self._name = name
-
- def iterdir(self):
- return iter(())
-
- def is_file(self):
- return self._reader.is_resource(self.name)
-
- def is_dir(self):
- return not self.is_file()
-
- def joinpath(self, other):
- return CompatibilityFiles.OrphanPath(self.name, other)
-
- @property
- def name(self):
- return self._name
-
- def open(self, mode='r', *args, **kwargs):
- return _io_wrapper(
- self._reader.open_resource(self.name), mode, *args, **kwargs
- )
-
- class OrphanPath(abc.Traversable):
- """
- Orphan path, not tied to a module spec or resource reader.
- Can't be read and doesn't expose any meaningful children.
- """
-
- def __init__(self, *path_parts):
- if len(path_parts) < 1:
- raise ValueError('Need at least one path part to construct a path')
- self._path = path_parts
-
- def iterdir(self):
- return iter(())
-
- def is_file(self):
- return False
-
- is_dir = is_file
-
- def joinpath(self, other):
- return CompatibilityFiles.OrphanPath(*self._path, other)
-
- @property
- def name(self):
- return self._path[-1]
-
- def open(self, mode='r', *args, **kwargs):
- raise FileNotFoundError("Can't open orphan path")
-
- def __init__(self, spec):
- self.spec = spec
-
- @property
- def _reader(self):
- with suppress(AttributeError):
- return self.spec.loader.get_resource_reader(self.spec.name)
-
- def _native(self):
- """
- Return the native reader if it supports files().
- """
- reader = self._reader
- return reader if hasattr(reader, 'files') else self
-
- def __getattr__(self, attr):
- return getattr(self._reader, attr)
-
- def files(self):
- return CompatibilityFiles.SpecPath(self.spec, self._reader)
-
-
-def wrap_spec(package):
- """
- Construct a package spec with traversable compatibility
- on the spec/loader/reader.
- """
- return SpecLoaderAdapter(package.__spec__, TraversableResourcesLoader)
diff --git a/contrib/tools/python3/src/Lib/importlib/resources/_common.py b/contrib/tools/python3/src/Lib/importlib/resources/_common.py
deleted file mode 100644
index a3902535342..00000000000
--- a/contrib/tools/python3/src/Lib/importlib/resources/_common.py
+++ /dev/null
@@ -1,207 +0,0 @@
-import os
-import pathlib
-import tempfile
-import functools
-import contextlib
-import types
-import importlib
-import inspect
-import warnings
-import itertools
-
-from typing import Union, Optional, cast
-from .abc import ResourceReader, Traversable
-
-from ._adapters import wrap_spec
-
-Package = Union[types.ModuleType, str]
-Anchor = Package
-
-
-def package_to_anchor(func):
- """
- 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
- """
- 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
-
-
-@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.
- """
- # We can't use
- # a issubclass() check here because apparently abc.'s __subclasscheck__()
- # hook wants to create a weak reference to the object, but
- # zipimport.zipimporter does not support weak references, resulting in a
- # TypeError. That seems terrible.
- spec = package.__spec__
- reader = getattr(spec.loader, 'get_resource_reader', None) # type: ignore
- if reader is None:
- return None
- return reader(spec.name) # type: ignore
-
-
-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 _infer_caller():
- """
- 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'
-
- 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.
-
- """
- spec = wrap_spec(package)
- reader = spec.loader.get_resource_reader(spec.name)
- return reader.files()
-
-
-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.
- fd, raw_path = tempfile.mkstemp(suffix=suffix)
- try:
- try:
- os.write(fd, reader())
- finally:
- os.close(fd)
- del reader
- yield pathlib.Path(raw_path)
- finally:
- try:
- _os_remove(raw_path)
- except FileNotFoundError:
- 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
-
-
-def as_file(path):
- """
- Given a Traversable object, return that object as a
- path on the local file system in a context manager.
- """
- return _temp_dir(path) if _is_present_dir(path) else _temp_file(path)
-
-
-@as_file.register(pathlib.Path)
-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
deleted file mode 100644
index 7b775ef5ae8..00000000000
--- a/contrib/tools/python3/src/Lib/importlib/resources/_itertools.py
+++ /dev/null
@@ -1,38 +0,0 @@
-# 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)
-
- try:
- second_value = next(it)
- except StopIteration:
- pass
- else:
- 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
deleted file mode 100644
index b1ea8105dad..00000000000
--- a/contrib/tools/python3/src/Lib/importlib/resources/_legacy.py
+++ /dev/null
@@ -1,120 +0,0 @@
-import functools
-import os
-import pathlib
-import types
-import warnings
-
-from typing import Union, Iterable, ContextManager, BinaryIO, TextIO, Any
-
-from . import _common
-
-Package = Union[types.ModuleType, str]
-Resource = str
-
-
-def deprecated(func):
- @functools.wraps(func)
- def wrapper(*args, **kwargs):
- warnings.warn(
- f"{func.__name__} is deprecated. Use files() instead. "
- "Refer to https://importlib-resources.readthedocs.io"
- "/en/latest/using.html#migrating-from-legacy for migration advice.",
- DeprecationWarning,
- stacklevel=2,
- )
- return func(*args, **kwargs)
-
- return wrapper
-
-
-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.
- """
- str_path = str(path)
- parent, file_name = os.path.split(str_path)
- if parent:
- raise ValueError(f'{path!r} must be only a file name')
- return file_name
-
-
-@deprecated
-def open_binary(package: Package, resource: Resource) -> BinaryIO:
- """Return a file-like object opened for binary reading of the resource."""
- return (_common.files(package) / normalize_path(resource)).open('rb')
-
-
-@deprecated
-def read_binary(package: Package, resource: Resource) -> bytes:
- """Return the binary contents of the resource."""
- return (_common.files(package) / normalize_path(resource)).read_bytes()
-
-
-@deprecated
-def open_text(
- package: Package,
- resource: Resource,
- encoding: str = 'utf-8',
- errors: str = 'strict',
-) -> TextIO:
- """Return a file-like object opened for text reading of the resource."""
- return (_common.files(package) / normalize_path(resource)).open(
- 'r', encoding=encoding, errors=errors
- )
-
-
-@deprecated
-def read_text(
- package: Package,
- resource: Resource,
- encoding: str = 'utf-8',
- errors: str = 'strict',
-) -> str:
- """Return the decoded string of the resource.
-
- The decoding-related arguments have the same semantics as those of
- bytes.decode().
- """
- with open_text(package, resource, encoding, errors) as fp:
- return fp.read()
-
-
-@deprecated
-def contents(package: Package) -> Iterable[str]:
- """Return an iterable of entries in `package`.
-
- Note that not all entries are resources. Specifically, directories are
- not considered resources. Use `is_resource()` on each entry returned here
- to check if it is a resource or not.
- """
- return [path.name for path in _common.files(package).iterdir()]
-
-
-@deprecated
-def is_resource(package: Package, name: str) -> bool:
- """True if `name` is a resource inside `package`.
-
- Directories are *not* resources.
- """
- resource = normalize_path(name)
- return any(
- traversable.name == resource and traversable.is_file()
- for traversable in _common.files(package).iterdir()
- )
-
-
-@deprecated
-def path(
- package: Package,
- resource: Resource,
-) -> ContextManager[pathlib.Path]:
- """A context manager providing a file path object to the resource.
-
- If the resource does not already exist on its own on the file system,
- a temporary file will be created. If the file was created, the file
- will be deleted upon exiting the context manager (no exception is
- raised if the file was deleted prior to the context manager
- exiting).
- """
- return _common.as_file(_common.files(package) / normalize_path(resource))
diff --git a/contrib/tools/python3/src/Lib/importlib/resources/abc.py b/contrib/tools/python3/src/Lib/importlib/resources/abc.py
deleted file mode 100644
index 6750a7aaf14..00000000000
--- a/contrib/tools/python3/src/Lib/importlib/resources/abc.py
+++ /dev/null
@@ -1,173 +0,0 @@
-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
-
-
-StrPath = Union[str, os.PathLike[str]]
-
-__all__ = ["ResourceReader", "Traversable", "TraversableResources"]
-
-
-class ResourceReader(metaclass=abc.ABCMeta):
- """Abstract base class for loaders to provide resource reading support."""
-
- @abc.abstractmethod
- def open_resource(self, resource: Text) -> BinaryIO:
- """Return an opened, file-like object for binary reading.
-
- The 'resource' argument is expected to represent only a file name.
- If the resource cannot be found, FileNotFoundError is raised.
- """
- # This deliberately raises FileNotFoundError instead of
- # NotImplementedError so that if this method is accidentally called,
- # it'll still do the right thing.
- raise FileNotFoundError
-
- @abc.abstractmethod
- def resource_path(self, resource: Text) -> Text:
- """Return the file system path to the specified resource.
-
- The 'resource' argument is expected to represent only a file name.
- If the resource does not exist on the file system, raise
- FileNotFoundError.
- """
- # This deliberately raises FileNotFoundError instead of
- # NotImplementedError so that if this method is accidentally called,
- # it'll still do the right thing.
- raise FileNotFoundError
-
- @abc.abstractmethod
- def is_resource(self, path: Text) -> bool:
- """Return True if the named 'path' is a resource.
-
- Files are resources, directories are not.
- """
- raise FileNotFoundError
-
- @abc.abstractmethod
- def contents(self) -> Iterable[str]:
- """Return an iterable of entries in `package`."""
- raise FileNotFoundError
-
-
-class TraversalError(Exception):
- pass
-
-
-@runtime_checkable
-class Traversable(Protocol):
- """
- An object with a subset of pathlib.Path methods suitable for
- traversing directories and opening files.
-
- Any exceptions that occur when accessing the backing resource
- may propagate unaltered.
- """
-
- @abc.abstractmethod
- def iterdir(self) -> Iterator["Traversable"]:
- """
- Yield Traversable objects in self
- """
-
- def read_bytes(self) -> bytes:
- """
- Read contents of self as bytes
- """
- with self.open('rb') as strm:
- return strm.read()
-
- def read_text(self, encoding: Optional[str] = None) -> str:
- """
- Read contents of self as text
- """
- with self.open(encoding=encoding) as strm:
- return strm.read()
-
- @abc.abstractmethod
- def is_dir(self) -> bool:
- """
- Return True if self is a directory
- """
-
- @abc.abstractmethod
- def is_file(self) -> bool:
- """
- Return True if self is a file
- """
-
- def joinpath(self, *descendants: StrPath) -> "Traversable":
- """
- Return Traversable resolved with any descendants applied.
-
- Each descendant should be a path segment relative to self
- 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":
- """
- Return Traversable child in self
- """
- return self.joinpath(child)
-
- @abc.abstractmethod
- def open(self, mode='r', *args, **kwargs):
- """
- mode may be 'r' or 'rb' to open as text or binary. Return a handle
- suitable for reading (same as pathlib.Path.open).
-
- When opening as text, accepts encoding parameters such as those
- accepted by io.TextIOWrapper.
- """
-
- @property
- @abc.abstractmethod
- def name(self) -> str:
- """
- The base name of this object without any parent references.
- """
-
-
-class TraversableResources(ResourceReader):
- """
- The required interface for providing traversable
- resources.
- """
-
- @abc.abstractmethod
- def files(self) -> "Traversable":
- """Return a Traversable object for the loaded package."""
-
- def open_resource(self, resource: StrPath) -> io.BufferedReader:
- return self.files().joinpath(resource).open('rb')
-
- def resource_path(self, resource: Any) -> NoReturn:
- raise FileNotFoundError(resource)
-
- def is_resource(self, path: StrPath) -> bool:
- return self.files().joinpath(path).is_file()
-
- def contents(self) -> Iterator[str]:
- return (item.name for item in self.files().iterdir())
diff --git a/contrib/tools/python3/src/Lib/importlib/resources/readers.py b/contrib/tools/python3/src/Lib/importlib/resources/readers.py
deleted file mode 100644
index c3cdf769cbe..00000000000
--- a/contrib/tools/python3/src/Lib/importlib/resources/readers.py
+++ /dev/null
@@ -1,144 +0,0 @@
-import collections
-import itertools
-import pathlib
-import operator
-import zipfile
-
-from . import abc
-
-from ._itertools import only
-
-
-def remove_duplicates(items):
- return iter(collections.OrderedDict.fromkeys(items))
-
-
-class FileReader(abc.TraversableResources):
- def __init__(self, loader):
- self.path = pathlib.Path(loader.path).parent
-
- def resource_path(self, resource):
- """
- Return the file system path to prevent
- `resources.path()` from creating a temporary
- copy.
- """
- return str(self.path.joinpath(resource))
-
- def files(self):
- return self.path
-
-
-class ZipReader(abc.TraversableResources):
- def __init__(self, loader, module):
- _, _, name = module.rpartition('.')
- self.prefix = loader.prefix.replace('\\', '/') + name + '/'
- self.archive = loader.archive
-
- def open_resource(self, resource):
- try:
- return super().open_resource(resource)
- except KeyError as exc:
- raise FileNotFoundError(exc.args[0])
-
- def is_resource(self, path):
- """
- Workaround for `zipfile.Path.is_file` returning true
- for non-existent paths.
- """
- target = self.files().joinpath(path)
- return target.is_file() and target.exists()
-
- def files(self):
- return zipfile.Path(self.archive, self.prefix)
-
-
-class MultiplexedPath(abc.Traversable):
- """
- Given a series of Traversable objects, implement a merged
- version of the interface across all objects. Useful for
- namespace packages which may be multihomed at a single
- name.
- """
-
- def __init__(self, *paths):
- self._paths = list(map(pathlib.Path, remove_duplicates(paths)))
- if not self._paths:
- message = 'MultiplexedPath must contain at least one path'
- raise FileNotFoundError(message)
- if not all(path.is_dir() for path in self._paths):
- raise NotADirectoryError('MultiplexedPath only supports directories')
-
- def iterdir(self):
- 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')
-
- def read_text(self, *args, **kwargs):
- raise FileNotFoundError(f'{self} is not a file')
-
- def is_dir(self):
- return True
-
- def is_file(self):
- return False
-
- 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.
-
- 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')
-
- @property
- def name(self):
- return self._paths[0].name
-
- def __repr__(self):
- paths = ', '.join(f"'{path}'" for path in self._paths)
- return f'MultiplexedPath({paths})'
-
-
-class NamespaceReader(abc.TraversableResources):
- def __init__(self, namespace_path):
- if 'NamespacePath' not in str(namespace_path):
- raise ValueError('Invalid path')
- self.path = MultiplexedPath(*list(namespace_path))
-
- def resource_path(self, resource):
- """
- Return the file system path to prevent
- `resources.path()` from creating a temporary
- copy.
- """
- return str(self.path.joinpath(resource))
-
- def files(self):
- return self.path
diff --git a/contrib/tools/python3/src/Lib/importlib/resources/simple.py b/contrib/tools/python3/src/Lib/importlib/resources/simple.py
deleted file mode 100644
index 7770c922c84..00000000000
--- a/contrib/tools/python3/src/Lib/importlib/resources/simple.py
+++ /dev/null
@@ -1,106 +0,0 @@
-"""
-Interface adapters for low-level readers.
-"""
-
-import abc
-import io
-import itertools
-from typing import BinaryIO, List
-
-from .abc import Traversable, TraversableResources
-
-
-class SimpleReader(abc.ABC):
- """
- The minimum, low-level interface required from a resource
- provider.
- """
-
- @property
- @abc.abstractmethod
- def package(self) -> str:
- """
- The name of the package for which this reader loads resources.
- """
-
- @abc.abstractmethod
- def children(self) -> List['SimpleReader']:
- """
- Obtain an iterable of SimpleReader for available
- child containers (e.g. directories).
- """
-
- @abc.abstractmethod
- def resources(self) -> List[str]:
- """
- Obtain available named resources for this virtual package.
- """
-
- @abc.abstractmethod
- def open_binary(self, resource: str) -> BinaryIO:
- """
- Obtain a File-like for a named resource.
- """
-
- @property
- def name(self):
- return self.package.split('.')[-1]
-
-
-class ResourceContainer(Traversable):
- """
- Traversable container for a package's resources via its reader.
- """
-
- def __init__(self, reader: SimpleReader):
- self.reader = reader
-
- def is_dir(self):
- return True
-
- def is_file(self):
- return False
-
- def iterdir(self):
- files = (ResourceHandle(self, name) for name in self.reader.resources)
- dirs = map(ResourceContainer, self.reader.children())
- return itertools.chain(files, dirs)
-
- def open(self, *args, **kwargs):
- raise IsADirectoryError()
-
-
-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):
- """
- A TraversableResources based on SimpleReader. Resource providers
- may derive from this class to provide the TraversableResources
- interface by supplying the SimpleReader interface.
- """
-
- def files(self):
- return ResourceContainer(self)