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