Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

bpo-46118: Move importlib.resources to its own package. #30176

Merged
merged 3 commits into from
Dec 31, 2021
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Jump to
Jump to file
Failed to load files.
Diff view
Diff view
148 changes: 13 additions & 135 deletions Lib/importlib/abc.py
Original file line number Diff line number Diff line change
Expand Up @@ -14,8 +14,19 @@
from ._abc import Loader
import abc
import warnings
from typing import BinaryIO, Iterable, Text
from typing import Protocol, runtime_checkable

# for compatibility with Python 3.10
from .resources.abc import ResourceReader, Traversable, TraversableResources


__all__ = [
'Loader', 'Finder', 'MetaPathFinder', 'PathEntryFinder',
'ResourceLoader', 'InspectLoader', 'ExecutionLoader',
'FileLoader', 'SourceLoader',

# for compatibility with Python 3.10
'ResourceReader', 'Traversable', 'TraversableResources',
]


def _register(abstract_cls, *classes):
Expand Down Expand Up @@ -307,136 +318,3 @@ def set_data(self, path, data):
"""

_register(SourceLoader, machinery.SourceFileLoader)


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


@runtime_checkable
class Traversable(Protocol):
"""
An object with a subset of pathlib.Path methods suitable for
traversing directories and opening files.
"""

@abc.abstractmethod
def iterdir(self):
"""
Yield Traversable objects in self
"""

def read_bytes(self):
"""
Read contents of self as bytes
"""
with self.open('rb') as strm:
return strm.read()

def read_text(self, encoding=None):
"""
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
"""

@abc.abstractmethod
def joinpath(self, child):
"""
Return Traversable child in self
"""

def __truediv__(self, child):
"""
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.
"""

@abc.abstractproperty
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):
"""Return a Traversable object for the loaded package."""

def open_resource(self, resource):
return self.files().joinpath(resource).open('rb')

def resource_path(self, resource):
raise FileNotFoundError(resource)

def is_resource(self, path):
return self.files().joinpath(path).is_file()

def contents(self):
return (item.name for item in self.files().iterdir())
128 changes: 9 additions & 119 deletions Lib/importlib/readers.py
Original file line number Diff line number Diff line change
@@ -1,122 +1,12 @@
import collections
import operator
import pathlib
import zipfile
"""
Compatibility shim for .resources.readers as found on Python 3.10.

from . import abc
Consumers that can rely on Python 3.11 should use the other
module directly.
"""

from ._itertools import unique_everseen
from .resources.readers import (
FileReader, ZipReader, MultiplexedPath, NamespaceReader,
)


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):
files = (file for path in self._paths for file in path.iterdir())
return unique_everseen(files, key=operator.attrgetter('name'))

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, 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 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
__all__ = ['FileReader', 'ZipReader', 'MultiplexedPath', 'NamespaceReader']
Original file line number Diff line number Diff line change
Expand Up @@ -17,7 +17,7 @@
Resource,
)

from importlib.abc import ResourceReader
from .abc import ResourceReader


__all__ = [
Expand Down
File renamed without changes.
File renamed without changes.
File renamed without changes.
File renamed without changes.