1 import abc
2 import io
3 import itertools
4 import os
5 import pathlib
6 from typing import Any, BinaryIO, Iterable, Iterator, NoReturn, Text, Optional
7 from typing import runtime_checkable, Protocol
8 from typing import Union
9
10
11 StrPath = Union[str, os.PathLike[str]]
12
13 __all__ = ["ResourceReader", "Traversable", "TraversableResources"]
14
15
16 class ESC[4;38;5;81mResourceReader(metaclass=ESC[4;38;5;149mabcESC[4;38;5;149m.ESC[4;38;5;149mABCMeta):
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
58 class ESC[4;38;5;81mTraversalError(ESC[4;38;5;149mException):
59 pass
60
61
62 @runtime_checkable
63 class ESC[4;38;5;81mTraversable(ESC[4;38;5;149mProtocol):
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
153 class ESC[4;38;5;81mTraversableResources(ESC[4;38;5;149mResourceReader):
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())