• Home
  • Line#
  • Scopes#
  • Navigate#
  • Raw
  • Download
1import abc
2import io
3import itertools
4import os
5import pathlib
6from typing import Any, BinaryIO, Iterable, Iterator, NoReturn, Text, Optional
7from typing import runtime_checkable, Protocol
8from typing import Union
9
10
11StrPath = Union[str, os.PathLike[str]]
12
13__all__ = ["ResourceReader", "Traversable", "TraversableResources"]
14
15
16class ResourceReader(metaclass=abc.ABCMeta):
17    """Abstract base class for loaders to provide resource reading support."""
18
19    @abc.abstractmethod
20    def open_resource(self, resource: Text) -> BinaryIO:
21        """Return an opened, file-like object for binary reading.
22
23        The 'resource' argument is expected to represent only a file name.
24        If the resource cannot be found, FileNotFoundError is raised.
25        """
26        # This deliberately raises FileNotFoundError instead of
27        # NotImplementedError so that if this method is accidentally called,
28        # it'll still do the right thing.
29        raise FileNotFoundError
30
31    @abc.abstractmethod
32    def resource_path(self, resource: Text) -> Text:
33        """Return the file system path to the specified resource.
34
35        The 'resource' argument is expected to represent only a file name.
36        If the resource does not exist on the file system, raise
37        FileNotFoundError.
38        """
39        # This deliberately raises FileNotFoundError instead of
40        # NotImplementedError so that if this method is accidentally called,
41        # it'll still do the right thing.
42        raise FileNotFoundError
43
44    @abc.abstractmethod
45    def is_resource(self, path: Text) -> bool:
46        """Return True if the named 'path' is a resource.
47
48        Files are resources, directories are not.
49        """
50        raise FileNotFoundError
51
52    @abc.abstractmethod
53    def contents(self) -> Iterable[str]:
54        """Return an iterable of entries in `package`."""
55        raise FileNotFoundError
56
57
58class TraversalError(Exception):
59    pass
60
61
62@runtime_checkable
63class Traversable(Protocol):
64    """
65    An object with a subset of pathlib.Path methods suitable for
66    traversing directories and opening files.
67
68    Any exceptions that occur when accessing the backing resource
69    may propagate unaltered.
70    """
71
72    @abc.abstractmethod
73    def iterdir(self) -> Iterator["Traversable"]:
74        """
75        Yield Traversable objects in self
76        """
77
78    def read_bytes(self) -> bytes:
79        """
80        Read contents of self as bytes
81        """
82        with self.open('rb') as strm:
83            return strm.read()
84
85    def read_text(self, encoding: Optional[str] = None) -> str:
86        """
87        Read contents of self as text
88        """
89        with self.open(encoding=encoding) as strm:
90            return strm.read()
91
92    @abc.abstractmethod
93    def is_dir(self) -> bool:
94        """
95        Return True if self is a directory
96        """
97
98    @abc.abstractmethod
99    def is_file(self) -> bool:
100        """
101        Return True if self is a file
102        """
103
104    def joinpath(self, *descendants: StrPath) -> "Traversable":
105        """
106        Return Traversable resolved with any descendants applied.
107
108        Each descendant should be a path segment relative to self
109        and each may contain multiple levels separated by
110        ``posixpath.sep`` (``/``).
111        """
112        if not descendants:
113            return self
114        names = itertools.chain.from_iterable(
115            path.parts for path in map(pathlib.PurePosixPath, descendants)
116        )
117        target = next(names)
118        matches = (
119            traversable for traversable in self.iterdir() if traversable.name == target
120        )
121        try:
122            match = next(matches)
123        except StopIteration:
124            raise TraversalError(
125                "Target not found during traversal.", target, list(names)
126            )
127        return match.joinpath(*names)
128
129    def __truediv__(self, child: StrPath) -> "Traversable":
130        """
131        Return Traversable child in self
132        """
133        return self.joinpath(child)
134
135    @abc.abstractmethod
136    def open(self, mode='r', *args, **kwargs):
137        """
138        mode may be 'r' or 'rb' to open as text or binary. Return a handle
139        suitable for reading (same as pathlib.Path.open).
140
141        When opening as text, accepts encoding parameters such as those
142        accepted by io.TextIOWrapper.
143        """
144
145    @property
146    @abc.abstractmethod
147    def name(self) -> str:
148        """
149        The base name of this object without any parent references.
150        """
151
152
153class TraversableResources(ResourceReader):
154    """
155    The required interface for providing traversable
156    resources.
157    """
158
159    @abc.abstractmethod
160    def files(self) -> "Traversable":
161        """Return a Traversable object for the loaded package."""
162
163    def open_resource(self, resource: StrPath) -> io.BufferedReader:
164        return self.files().joinpath(resource).open('rb')
165
166    def resource_path(self, resource: Any) -> NoReturn:
167        raise FileNotFoundError(resource)
168
169    def is_resource(self, path: StrPath) -> bool:
170        return self.files().joinpath(path).is_file()
171
172    def contents(self) -> Iterator[str]:
173        return (item.name for item in self.files().iterdir())
174