diff options
Diffstat (limited to 'Lib/importlib/resources.py')
-rw-r--r-- | Lib/importlib/resources.py | 259 |
1 files changed, 0 insertions, 259 deletions
diff --git a/Lib/importlib/resources.py b/Lib/importlib/resources.py deleted file mode 100644 index fc3a1c9..0000000 --- a/Lib/importlib/resources.py +++ /dev/null @@ -1,259 +0,0 @@ -import os -import tempfile - -from . import abc as resources_abc -from contextlib import contextmanager, suppress -from importlib import import_module -from importlib.abc import ResourceLoader -from io import BytesIO, TextIOWrapper -from pathlib import Path -from types import ModuleType -from typing import Iterable, Iterator, Optional, Set, Union # noqa: F401 -from typing import cast -from typing.io import BinaryIO, TextIO -from zipimport import ZipImportError - - -__all__ = [ - 'Package', - 'Resource', - 'contents', - 'is_resource', - 'open_binary', - 'open_text', - 'path', - 'read_binary', - 'read_text', - ] - - -Package = Union[str, ModuleType] -Resource = Union[str, os.PathLike] - - -def _get_package(package) -> ModuleType: - """Take a package name or module object and return the module. - - If a name, the module is imported. If the passed or imported module - object is not a package, raise an exception. - """ - if hasattr(package, '__spec__'): - if package.__spec__.submodule_search_locations is None: - raise TypeError('{!r} is not a package'.format( - package.__spec__.name)) - else: - return package - else: - module = import_module(package) - if module.__spec__.submodule_search_locations is None: - raise TypeError('{!r} is not a package'.format(package)) - else: - return module - - -def _normalize_path(path) -> str: - """Normalize a path by ensuring it is a string. - - If the resulting string contains path separators, an exception is raised. - """ - parent, file_name = os.path.split(path) - if parent: - raise ValueError('{!r} must be only a file name'.format(path)) - else: - return file_name - - -def _get_resource_reader( - package: ModuleType) -> Optional[resources_abc.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__ - if hasattr(spec.loader, 'get_resource_reader'): - return cast(resources_abc.ResourceReader, - spec.loader.get_resource_reader(spec.name)) - return None - - -def _check_location(package): - if package.__spec__.origin is None or not package.__spec__.has_location: - raise FileNotFoundError(f'Package has no location {package!r}') - - -def open_binary(package: Package, resource: Resource) -> BinaryIO: - """Return a file-like object opened for binary reading of the resource.""" - resource = _normalize_path(resource) - package = _get_package(package) - reader = _get_resource_reader(package) - if reader is not None: - return reader.open_resource(resource) - _check_location(package) - absolute_package_path = os.path.abspath(package.__spec__.origin) - package_path = os.path.dirname(absolute_package_path) - full_path = os.path.join(package_path, resource) - try: - return open(full_path, mode='rb') - except OSError: - # Just assume the loader is a resource loader; all the relevant - # importlib.machinery loaders are and an AttributeError for - # get_data() will make it clear what is needed from the loader. - loader = cast(ResourceLoader, package.__spec__.loader) - data = None - if hasattr(package.__spec__.loader, 'get_data'): - with suppress(OSError): - data = loader.get_data(full_path) - if data is None: - package_name = package.__spec__.name - message = '{!r} resource not found in {!r}'.format( - resource, package_name) - raise FileNotFoundError(message) - else: - return BytesIO(data) - - -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.""" - resource = _normalize_path(resource) - package = _get_package(package) - reader = _get_resource_reader(package) - if reader is not None: - return TextIOWrapper(reader.open_resource(resource), encoding, errors) - _check_location(package) - absolute_package_path = os.path.abspath(package.__spec__.origin) - package_path = os.path.dirname(absolute_package_path) - full_path = os.path.join(package_path, resource) - try: - return open(full_path, mode='r', encoding=encoding, errors=errors) - except OSError: - # Just assume the loader is a resource loader; all the relevant - # importlib.machinery loaders are and an AttributeError for - # get_data() will make it clear what is needed from the loader. - loader = cast(ResourceLoader, package.__spec__.loader) - data = None - if hasattr(package.__spec__.loader, 'get_data'): - with suppress(OSError): - data = loader.get_data(full_path) - if data is None: - package_name = package.__spec__.name - message = '{!r} resource not found in {!r}'.format( - resource, package_name) - raise FileNotFoundError(message) - else: - return TextIOWrapper(BytesIO(data), encoding, errors) - - -def read_binary(package: Package, resource: Resource) -> bytes: - """Return the binary contents of the resource.""" - resource = _normalize_path(resource) - package = _get_package(package) - with open_binary(package, resource) as fp: - return fp.read() - - -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(). - """ - resource = _normalize_path(resource) - package = _get_package(package) - with open_text(package, resource, encoding, errors) as fp: - return fp.read() - - -@contextmanager -def path(package: Package, resource: Resource) -> Iterator[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). - """ - resource = _normalize_path(resource) - package = _get_package(package) - reader = _get_resource_reader(package) - if reader is not None: - try: - yield Path(reader.resource_path(resource)) - return - except FileNotFoundError: - pass - else: - _check_location(package) - # Fall-through for both the lack of resource_path() *and* if - # resource_path() raises FileNotFoundError. - package_directory = Path(package.__spec__.origin).parent - file_path = package_directory / resource - if file_path.exists(): - yield file_path - else: - with open_binary(package, resource) as fp: - data = fp.read() - # 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() - try: - os.write(fd, data) - os.close(fd) - yield Path(raw_path) - finally: - try: - os.remove(raw_path) - except FileNotFoundError: - pass - - -def is_resource(package: Package, name: str) -> bool: - """True if 'name' is a resource inside 'package'. - - Directories are *not* resources. - """ - package = _get_package(package) - _normalize_path(name) - reader = _get_resource_reader(package) - if reader is not None: - return reader.is_resource(name) - try: - package_contents = set(contents(package)) - except (NotADirectoryError, FileNotFoundError): - return False - if name not in package_contents: - return False - # Just because the given file_name lives as an entry in the package's - # contents doesn't necessarily mean it's a resource. Directories are not - # resources, so let's try to find out if it's a directory or not. - path = Path(package.__spec__.origin).parent / name - return path.is_file() - - -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. - """ - package = _get_package(package) - reader = _get_resource_reader(package) - if reader is not None: - return reader.contents() - # Is the package a namespace package? By definition, namespace packages - # cannot have resources. We could use _check_location() and catch the - # exception, but that's extra work, so just inline the check. - elif package.__spec__.origin is None or not package.__spec__.has_location: - return () - else: - package_directory = Path(package.__spec__.origin).parent - return os.listdir(package_directory) |