Server IP : 85.214.239.14 / Your IP : 18.117.101.250 Web Server : Apache/2.4.62 (Debian) System : Linux h2886529.stratoserver.net 4.9.0 #1 SMP Tue Jan 9 19:45:01 MSK 2024 x86_64 User : www-data ( 33) PHP Version : 7.4.18 Disable Function : pcntl_alarm,pcntl_fork,pcntl_waitpid,pcntl_wait,pcntl_wifexited,pcntl_wifstopped,pcntl_wifsignaled,pcntl_wifcontinued,pcntl_wexitstatus,pcntl_wtermsig,pcntl_wstopsig,pcntl_signal,pcntl_signal_get_handler,pcntl_signal_dispatch,pcntl_get_last_error,pcntl_strerror,pcntl_sigprocmask,pcntl_sigwaitinfo,pcntl_sigtimedwait,pcntl_exec,pcntl_getpriority,pcntl_setpriority,pcntl_async_signals,pcntl_unshare, MySQL : OFF | cURL : OFF | WGET : ON | Perl : ON | Python : ON | Sudo : ON | Pkexec : OFF Directory : /usr/lib/python3.11/importlib/resources/ |
Upload File : |
""" 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. """ @abc.abstractproperty def package(self): # type: () -> str """ The name of the package for which this reader loads resources. """ @abc.abstractmethod def children(self): # type: () -> List['SimpleReader'] """ Obtain an iterable of SimpleReader for available child containers (e.g. directories). """ @abc.abstractmethod def resources(self): # type: () -> List[str] """ Obtain available named resources for this virtual package. """ @abc.abstractmethod def open_binary(self, resource): # type: (str) -> BinaryIO """ Obtain a File-like for a named resource. """ @property def name(self): return self.package.split('.')[-1] class ResourceHandle(Traversable): """ Handle to a named resource in a ResourceReader. """ def __init__(self, parent, name): # type: (ResourceContainer, str) -> None 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 ResourceContainer(Traversable): """ Traversable container for a package's resources via its reader. """ def __init__(self, reader): # type: (SimpleReader) -> None 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() @staticmethod def _flatten(compound_names): for name in compound_names: yield from name.split('/') def joinpath(self, *descendants): if not descendants: return self names = self._flatten(descendants) target = next(names) return next( traversable for traversable in self.iterdir() if traversable.name == target ).joinpath(*names) 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)