mirrors/cpython
1
0
Fork 0
mirror of https://github.com/python/cpython.git synced 2025-11-02 11:08:57 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 14

bpo-42129: Add support for resources in namespaces (GH-24670)

Browse source 
* Unify behavior in ResourceReaderDefaultsTests and align with the behavior found in importlib_resources.
* Equip NamespaceLoader with a NamespaceReader.
* Apply changes from importlib_resources 5.0.4
This commit is contained in:
Jason R. Coombs 2021-03-04 13:43:00 -05:00 • committed by GitHub
parent fbf75b9997
commit 6714825414
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
21 changed files with 1315 additions and 916 deletions
Download patch file Download diff file Expand all files Collapse all files

106
Lib/importlib/resources.py
View file

@ -3,8 +3,10 @@ import io
from . import _common
from ._common import as_file, files
from .abc import ResourceReader
from contextlib import suppress
from importlib.abc import ResourceLoader
from importlib.machinery import ModuleSpec
from io import BytesIO, TextIOWrapper
from pathlib import Path
from types import ModuleType
@ -18,6 +20,7 @@ from functools import singledispatch
__all__ = [
'Package',
'Resource',
'ResourceReader',
'as_file',
'contents',
'files',
@ -27,7 +30,7 @@ __all__ = [
'path',
'read_binary',
'read_text',
]
]
Package = Union[str, ModuleType]
@ -41,36 +44,45 @@ def open_binary(package: Package, resource: Resource) -> BinaryIO:
reader = _common.get_resource_reader(package)
if reader is not None:
return reader.open_resource(resource)
absolute_package_path = os.path.abspath(
package.__spec__.origin or 'non-existent file')
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)
return BytesIO(data)
spec = cast(ModuleSpec, package.__spec__)
# Using pathlib doesn't work well here due to the lack of 'strict'
# argument for pathlib.Path.resolve() prior to Python 3.6.
if spec.submodule_search_locations is not None:
paths = spec.submodule_search_locations
elif spec.origin is not None:
paths = [os.path.dirname(os.path.abspath(spec.origin))]
for package_path in paths:
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, spec.loader)
data = None
if hasattr(spec.loader, 'get_data'):
with suppress(OSError):
data = loader.get_data(full_path)
if data is not None:
return BytesIO(data)
raise FileNotFoundError(
'{!r} resource not found in {!r}'.format(resource, spec.name)
)
def open_text(package: Package,
resource: Resource,
encoding: str = 'utf-8',
errors: str = 'strict') -> TextIO:
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 TextIOWrapper(
open_binary(package, resource), encoding=encoding, errors=errors)
open_binary(package, resource), encoding=encoding, errors=errors
)
def read_binary(package: Package, resource: Resource) -> bytes:
@ -79,10 +91,12 @@ def read_binary(package: Package, resource: Resource) -> bytes:
return fp.read()
def read_text(package: Package,
resource: Resource,
encoding: str = 'utf-8',
errors: str = 'strict') -> str:
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
@ -93,8 +107,9 @@ def read_text(package: Package,
def path(
package: Package, resource: Resource,
) -> 'ContextManager[Path]':
package: Package,
resource: Resource,
) -> 'ContextManager[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,
@ -106,15 +121,17 @@ def path(
reader = _common.get_resource_reader(_common.get_package(package))
return (
_path_from_reader(reader, _common.normalize_path(resource))
if reader else
_common.as_file(
_common.files(package).joinpath(_common.normalize_path(resource)))
if reader
else _common.as_file(
_common.files(package).joinpath(_common.normalize_path(resource))
)
)
def _path_from_reader(reader, resource):
return _path_from_resource_path(reader, resource) or \
_path_from_open_resource(reader, resource)
return _path_from_resource_path(reader, resource) or _path_from_open_resource(
reader, resource
)
def _path_from_resource_path(reader, resource):
@ -154,15 +171,10 @@ def contents(package: Package) -> Iterable[str]:
reader = _common.get_resource_reader(package)
if reader is not None:
return _ensure_sequence(reader.contents())
# Is the package a namespace package? By definition, namespace packages
# cannot have resources.
namespace = (
package.__spec__.origin is None or
package.__spec__.origin == 'namespace'
)
if namespace or not package.__spec__.has_location:
return ()
return list(item.name for item in _common.from_package(package).iterdir())
transversable = _common.from_package(package)
if transversable.is_dir():
return list(item.name for item in transversable.iterdir())
return []
@singledispatch