git.reasoningtechnology.com Git - SubU/blob

   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 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
  53 class 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
  79 class 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     def joinpath(self, name):
 103         return next(
 104             traversable for traversable in self.iterdir() if traversable.name == name
 105         )
 106
 107
 108 class TraversableReader(TraversableResources, SimpleReader):
 109     """
 110     A TraversableResources based on SimpleReader. Resource providers
 111     may derive from this class to provide the TraversableResources
 112     interface by supplying the SimpleReader interface.
 113     """
 114
 115     def files(self):
 116         return ResourceContainer(self)