import abc
import io
import itertools
import os
import pathlib
from typing import Any, BinaryIO, Iterable, Iterator, NoReturn, Text, Optional
from typing import runtime_checkable, Protocol
from typing import Union
StrPath = Union[str, os.PathLike[str]]
__all__ = ["ResourceReader", "Traversable", "TraversableResources"]
class ResourceReader(metaclass=abc.ABCMeta):
@abc.abstractmethod
def open_resource(self, resource: Text) -> BinaryIO:
raise FileNotFoundError
@abc.abstractmethod
def resource_path(self, resource: Text) -> Text:
raise FileNotFoundError
@abc.abstractmethod
def is_resource(self, path: Text) -> bool:
raise FileNotFoundError
@abc.abstractmethod
def contents(self) -> Iterable[str]:
raise FileNotFoundError
class TraversalError(Exception):
pass
@runtime_checkable
class Traversable(Protocol):
@abc.abstractmethod
def iterdir(self) -> Iterator["Traversable"]:
def read_bytes(self) -> bytes:
with self.open('rb') as strm:
return strm.read()
def read_text(self, encoding: Optional[str] = None) -> str:
with self.open(encoding=encoding) as strm:
return strm.read()
@abc.abstractmethod
def is_dir(self) -> bool:
@abc.abstractmethod
def is_file(self) -> bool:
def joinpath(self, *descendants: StrPath) -> "Traversable":
if not descendants:
return self
names = itertools.chain.from_iterable(
path.parts for path in map(pathlib.PurePosixPath, descendants)
)
target = next(names)
matches = (
traversable for traversable in self.iterdir() if traversable.name == target
)
try:
match = next(matches)
except StopIteration:
raise TraversalError(
"Target not found during traversal.", target, list(names)
)
return match.joinpath(*names)
def __truediv__(self, child: StrPath) -> "Traversable":
return self.joinpath(child)
@abc.abstractmethod
def open(self, mode='r', *args, **kwargs):
@property
@abc.abstractmethod
def name(self) -> str:
class TraversableResources(ResourceReader):
@abc.abstractmethod
def files(self) -> "Traversable":
def open_resource(self, resource: StrPath) -> io.BufferedReader:
return self.files().joinpath(resource).open('rb')
def resource_path(self, resource: Any) -> NoReturn:
raise FileNotFoundError(resource)
def is_resource(self, path: StrPath) -> bool:
return self.files().joinpath(path).is_file()
def contents(self) -> Iterator[str]:
return (item.name for item in self.files().iterdir())