codekingpro/portable-devtools

1import contextlib2import functools3import importlib4import inspect5import itertools6import os7import pathlib8import tempfile9import types10from typing import Optional, cast11 12from .abc import ResourceReader, Traversable13 14Package = types.ModuleType | str15Anchor = Package16 17 18def files(anchor: Optional[Anchor] = None) -> Traversable:19    """20    Get a Traversable resource for an anchor.21    """22    return from_package(resolve(anchor))23 24 25def get_resource_reader(package: types.ModuleType) -> Optional[ResourceReader]:26    """27    Return the package's loader if it's a ResourceReader.28    """29    # We can't use30    # a issubclass() check here because apparently abc.'s __subclasscheck__()31    # hook wants to create a weak reference to the object, but32    # zipimport.zipimporter does not support weak references, resulting in a33    # TypeError.  That seems terrible.34    spec = package.__spec__35    reader = getattr(spec.loader, 'get_resource_reader', None)  # type: ignore[union-attr]36    if reader is None:37        return None38    return reader(spec.name)  # type: ignore[union-attr]39 40 41@functools.singledispatch42def resolve(cand: Optional[Anchor]) -> types.ModuleType:43    return cast(types.ModuleType, cand)44 45 46@resolve.register47def _(cand: str) -> types.ModuleType:48    return importlib.import_module(cand)49 50 51@resolve.register52def _(cand: None) -> types.ModuleType:53    return resolve(_infer_caller().f_globals['__name__'])54 55 56def _infer_caller():57    """58    Walk the stack and find the frame of the first caller not in this module.59    """60 61    def is_this_file(frame_info):62        return frame_info.filename == stack[0].filename63 64    def is_wrapper(frame_info):65        return frame_info.function == 'wrapper'66 67    stack = inspect.stack()68    not_this_file = itertools.filterfalse(is_this_file, stack)69    # also exclude 'wrapper' due to singledispatch in the call stack70    callers = itertools.filterfalse(is_wrapper, not_this_file)71    return next(callers).frame72 73 74def _assert_spec(package: types.ModuleType) -> None:75    """76    Provide a nicer error message when package is ``__main__``77    and its ``__spec__`` is ``None``78    (https://docs.python.org/3/reference/import.html#main-spec).79    """80    if package.__spec__ is None:81        raise TypeError(82            f"Cannot access resources for '{package.__name__}' "83            "as it does not appear to correspond to an importable module (its __spec__ is None)."84        )85 86 87def from_package(package: types.ModuleType):88    """89    Return a Traversable object for the given package.90 91    """92    # deferred for performance (python/cpython#109829)93    from .future.adapters import wrap_spec94 95    _assert_spec(package)96    spec = wrap_spec(package)97    reader = spec.loader.get_resource_reader(spec.name)98    return reader.files()99 100 101@contextlib.contextmanager102def _tempfile(103    reader,104    suffix='',105    # gh-93353: Keep a reference to call os.remove() in late Python106    # finalization.107    *,108    _os_remove=os.remove,109):110    # Not using tempfile.NamedTemporaryFile as it leads to deeper 'try'111    # blocks due to the need to close the temporary file to work on Windows112    # properly.113    fd, raw_path = tempfile.mkstemp(suffix=suffix)114    try:115        try:116            os.write(fd, reader())117        finally:118            os.close(fd)119        del reader120        yield pathlib.Path(raw_path)121    finally:122        try:123            _os_remove(raw_path)124        except FileNotFoundError:125            pass126 127 128def _temp_file(path):129    return _tempfile(path.read_bytes, suffix=path.name)130 131 132def _is_present_dir(path: Traversable) -> bool:133    """134    Some Traversables implement ``is_dir()`` to raise an135    exception (i.e. ``FileNotFoundError``) when the136    directory doesn't exist. This function wraps that call137    to always return a boolean and only return True138    if there's a dir and it exists.139    """140    with contextlib.suppress(FileNotFoundError):141        return path.is_dir()142    return False143 144 145@functools.singledispatch146def as_file(path):147    """148    Given a Traversable object, return that object as a149    path on the local file system in a context manager.150    """151    return _temp_dir(path) if _is_present_dir(path) else _temp_file(path)152 153 154@as_file.register(pathlib.Path)155@contextlib.contextmanager156def _(path):157    """158    Degenerate behavior for pathlib.Path objects.159    """160    yield path161 162 163@contextlib.contextmanager164def _temp_path(dir: tempfile.TemporaryDirectory):165    """166    Wrap tempfile.TemporaryDirectory to return a pathlib object.167    """168    with dir as result:169        yield pathlib.Path(result)170 171 172@contextlib.contextmanager173def _temp_dir(path):174    """175    Given a traversable dir, recursively replicate the whole tree176    to the file system in a context manager.177    """178    assert path.is_dir()179    with _temp_path(tempfile.TemporaryDirectory()) as temp_dir:180        yield _write_contents(temp_dir, path)181 182 183def _write_contents(target, source):184    child = target.joinpath(source.name)185    if source.is_dir():186        child.mkdir()187        for item in source.iterdir():188            _write_contents(child, item)189    else:190        child.write_bytes(source.read_bytes())191    return child192