1 """
2 Interface adapters for low-level readers.
3 """
4
5 import abc
6 import io
7 import itertools
8 from typing import BinaryIO, List
9
10 from .abc import Traversable, TraversableResources
11
12
13 class ESC[4;38;5;81mSimpleReader(ESC[4;38;5;149mabcESC[4;38;5;149m.ESC[4;38;5;149mABC):
14 """
15 The minimum, low-level interface required from a resource
16 provider.
17 """
18
19 @property
20 @abc.abstractmethod
21 def package(self) -> str:
22 """
23 The name of the package for which this reader loads resources.
24 """
25
26 @abc.abstractmethod
27 def children(self) -> List['SimpleReader']:
28 """
29 Obtain an iterable of SimpleReader for available
30 child containers (e.g. directories).
31 """
32
33 @abc.abstractmethod
34 def resources(self) -> List[str]:
35 """
36 Obtain available named resources for this virtual package.
37 """
38
39 @abc.abstractmethod
40 def open_binary(self, resource: str) -> BinaryIO:
41 """
42 Obtain a File-like for a named resource.
43 """
44
45 @property
46 def name(self):
47 return self.package.split('.')[-1]
48
49
50 class ESC[4;38;5;81mResourceContainer(ESC[4;38;5;149mTraversable):
51 """
52 Traversable container for a package's resources via its reader.
53 """
54
55 def __init__(self, reader: SimpleReader):
56 self.reader = reader
57
58 def is_dir(self):
59 return True
60
61 def is_file(self):
62 return False
63
64 def iterdir(self):
65 files = (ResourceHandle(self, name) for name in self.reader.resources)
66 dirs = map(ResourceContainer, self.reader.children())
67 return itertools.chain(files, dirs)
68
69 def open(self, *args, **kwargs):
70 raise IsADirectoryError()
71
72
73 class ESC[4;38;5;81mResourceHandle(ESC[4;38;5;149mTraversable):
74 """
75 Handle to a named resource in a ResourceReader.
76 """
77
78 def __init__(self, parent: ResourceContainer, name: str):
79 self.parent = parent
80 self.name = name # type: ignore
81
82 def is_file(self):
83 return True
84
85 def is_dir(self):
86 return False
87
88 def open(self, mode='r', *args, **kwargs):
89 stream = self.parent.reader.open_binary(self.name)
90 if 'b' not in mode:
91 stream = io.TextIOWrapper(*args, **kwargs)
92 return stream
93
94 def joinpath(self, name):
95 raise RuntimeError("Cannot traverse into a resource")
96
97
98 class ESC[4;38;5;81mTraversableReader(ESC[4;38;5;149mTraversableResources, ESC[4;38;5;149mSimpleReader):
99 """
100 A TraversableResources based on SimpleReader. Resource providers
101 may derive from this class to provide the TraversableResources
102 interface by supplying the SimpleReader interface.
103 """
104
105 def files(self):
106 return ResourceContainer(self)