python (3.11.7)
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 @abc.abstractproperty
20 def package(self):
21 # type: () -> str
22 """
23 The name of the package for which this reader loads resources.
24 """
25
26 @abc.abstractmethod
27 def children(self):
28 # type: () -> List['SimpleReader']
29 """
30 Obtain an iterable of SimpleReader for available
31 child containers (e.g. directories).
32 """
33
34 @abc.abstractmethod
35 def resources(self):
36 # type: () -> List[str]
37 """
38 Obtain available named resources for this virtual package.
39 """
40
41 @abc.abstractmethod
42 def open_binary(self, resource):
43 # type: (str) -> BinaryIO
44 """
45 Obtain a File-like for a named resource.
46 """
47
48 @property
49 def name(self):
50 return self.package.split('.')[-1]
51
52
53 class ESC[4;38;5;81mResourceHandle(ESC[4;38;5;149mTraversable):
54 """
55 Handle to a named resource in a ResourceReader.
56 """
57
58 def __init__(self, parent, name):
59 # type: (ResourceContainer, str) -> None
60 self.parent = parent
61 self.name = name # type: ignore
62
63 def is_file(self):
64 return True
65
66 def is_dir(self):
67 return False
68
69 def open(self, mode='r', *args, **kwargs):
70 stream = self.parent.reader.open_binary(self.name)
71 if 'b' not in mode:
72 stream = io.TextIOWrapper(*args, **kwargs)
73 return stream
74
75 def joinpath(self, name):
76 raise RuntimeError("Cannot traverse into a resource")
77
78
79 class ESC[4;38;5;81mResourceContainer(ESC[4;38;5;149mTraversable):
80 """
81 Traversable container for a package's resources via its reader.
82 """
83
84 def __init__(self, reader):
85 # type: (SimpleReader) -> None
86 self.reader = reader
87
88 def is_dir(self):
89 return True
90
91 def is_file(self):
92 return False
93
94 def iterdir(self):
95 files = (ResourceHandle(self, name) for name in self.reader.resources)
96 dirs = map(ResourceContainer, self.reader.children())
97 return itertools.chain(files, dirs)
98
99 def open(self, *args, **kwargs):
100 raise IsADirectoryError()
101
102 @staticmethod
103 def _flatten(compound_names):
104 for name in compound_names:
105 yield from name.split('/')
106
107 def joinpath(self, *descendants):
108 if not descendants:
109 return self
110 names = self._flatten(descendants)
111 target = next(names)
112 return next(
113 traversable for traversable in self.iterdir() if traversable.name == target
114 ).joinpath(*names)
115
116
117 class ESC[4;38;5;81mTraversableReader(ESC[4;38;5;149mTraversableResources, ESC[4;38;5;149mSimpleReader):
118 """
119 A TraversableResources based on SimpleReader. Resource providers
120 may derive from this class to provide the TraversableResources
121 interface by supplying the SimpleReader interface.
122 """
123
124 def files(self):
125 return ResourceContainer(self)
Browse
Usage
Versions