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