mirror of
				https://github.com/python/cpython.git
				synced 2025-10-30 21:21:22 +00:00 
			
		
		
		
	
		
			
				
	
	
		
			173 lines
		
	
	
	
		
			5.1 KiB
		
	
	
	
		
			Python
		
	
	
	
	
	
			
		
		
	
	
			173 lines
		
	
	
	
		
			5.1 KiB
		
	
	
	
		
			Python
		
	
	
	
	
	
| import abc
 | |
| import io
 | |
| import itertools
 | |
| import os
 | |
| import pathlib
 | |
| from typing import Any, BinaryIO, Iterable, Iterator, NoReturn, Text, Optional
 | |
| from typing import runtime_checkable, Protocol
 | |
| from typing import Union
 | |
| 
 | |
| 
 | |
| StrPath = Union[str, os.PathLike[str]]
 | |
| 
 | |
| __all__ = ["ResourceReader", "Traversable", "TraversableResources"]
 | |
| 
 | |
| 
 | |
| 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
 | |
| 
 | |
| 
 | |
| class TraversalError(Exception):
 | |
|     pass
 | |
| 
 | |
| 
 | |
| @runtime_checkable
 | |
| class Traversable(Protocol):
 | |
|     """
 | |
|     An object with a subset of pathlib.Path methods suitable for
 | |
|     traversing directories and opening files.
 | |
| 
 | |
|     Any exceptions that occur when accessing the backing resource
 | |
|     may propagate unaltered.
 | |
|     """
 | |
| 
 | |
|     @abc.abstractmethod
 | |
|     def iterdir(self) -> Iterator["Traversable"]:
 | |
|         """
 | |
|         Yield Traversable objects in self
 | |
|         """
 | |
| 
 | |
|     def read_bytes(self) -> bytes:
 | |
|         """
 | |
|         Read contents of self as bytes
 | |
|         """
 | |
|         with self.open('rb') as strm:
 | |
|             return strm.read()
 | |
| 
 | |
|     def read_text(self, encoding: Optional[str] = None) -> str:
 | |
|         """
 | |
|         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
 | |
|         """
 | |
| 
 | |
|     def joinpath(self, *descendants: StrPath) -> "Traversable":
 | |
|         """
 | |
|         Return Traversable resolved with any descendants applied.
 | |
| 
 | |
|         Each descendant should be a path segment relative to self
 | |
|         and each may contain multiple levels separated by
 | |
|         ``posixpath.sep`` (``/``).
 | |
|         """
 | |
|         if not descendants:
 | |
|             return self
 | |
|         names = itertools.chain.from_iterable(
 | |
|             path.parts for path in map(pathlib.PurePosixPath, descendants)
 | |
|         )
 | |
|         target = next(names)
 | |
|         matches = (
 | |
|             traversable for traversable in self.iterdir() if traversable.name == target
 | |
|         )
 | |
|         try:
 | |
|             match = next(matches)
 | |
|         except StopIteration:
 | |
|             raise TraversalError(
 | |
|                 "Target not found during traversal.", target, list(names)
 | |
|             )
 | |
|         return match.joinpath(*names)
 | |
| 
 | |
|     def __truediv__(self, child: StrPath) -> "Traversable":
 | |
|         """
 | |
|         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.
 | |
|         """
 | |
| 
 | |
|     @property
 | |
|     @abc.abstractmethod
 | |
|     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) -> "Traversable":
 | |
|         """Return a Traversable object for the loaded package."""
 | |
| 
 | |
|     def open_resource(self, resource: StrPath) -> io.BufferedReader:
 | |
|         return self.files().joinpath(resource).open('rb')
 | |
| 
 | |
|     def resource_path(self, resource: Any) -> NoReturn:
 | |
|         raise FileNotFoundError(resource)
 | |
| 
 | |
|     def is_resource(self, path: StrPath) -> bool:
 | |
|         return self.files().joinpath(path).is_file()
 | |
| 
 | |
|     def contents(self) -> Iterator[str]:
 | |
|         return (item.name for item in self.files().iterdir())
 | 
