1""" 2Interface adapters for low-level readers. 3""" 4 5import abc 6import io 7import itertools 8from typing import BinaryIO, List 9 10from .abc import Traversable, TraversableResources 11 12 13class SimpleReader(abc.ABC): 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 53class ResourceHandle(Traversable): 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 79class ResourceContainer(Traversable): 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 117class TraversableReader(TraversableResources, SimpleReader): 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) 126