abc.py 5.1 KB

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