diff --git a/aws_doc_sdk_examples_tools/fs.py b/aws_doc_sdk_examples_tools/fs.py new file mode 100644 index 0000000..e980e60 --- /dev/null +++ b/aws_doc_sdk_examples_tools/fs.py @@ -0,0 +1,112 @@ +from abc import ABC, abstractmethod +from dataclasses import dataclass +from fnmatch import fnmatch +from os import listdir +from pathlib import Path +from stat import S_ISREG +from typing import Dict, Generator, List + + +@dataclass(frozen=True) +class Stat: + path: Path + exists: bool + is_file: bool + + @property + def is_dir(self): + return self.exists and not self.is_file + + +class Fs(ABC): + @abstractmethod + def glob(self, path: Path, glob: str) -> Generator[Path, None, None]: + pass + + @abstractmethod + def read(self, path: Path) -> str: + pass + + @abstractmethod + def write(self, path: Path, content: str): + pass + + @abstractmethod + def stat(self, path: Path) -> Stat: + pass + + @abstractmethod + def mkdir(self, path: Path): + pass + + @abstractmethod + def list(self, path: Path) -> List[Path]: + pass + + +class PathFs(Fs): + def glob(self, path: Path, glob: str) -> Generator[Path, None, None]: + return path.glob(glob) + + def read(self, path: Path) -> str: + with path.open("r") as file: + return file.read() + + def write(self, path: Path, content: str): + with path.open("w") as file: + file.write(content) + + def stat(self, path: Path) -> Stat: + if path.exists(): + stat = path.stat() + return Stat(path, True, S_ISREG(stat.st_mode)) + else: + return Stat(path, False, False) + + def mkdir(self, path: Path): + path.mkdir(parents=True, exist_ok=True) + + def list(self, path: Path) -> List[Path]: + if self.stat(path).is_file: + return [] + return [path / name for name in listdir(path)] + + +class RecordFs(Fs): + def __init__(self, fs: Dict[Path, str]): + self.fs = fs + + def glob(self, path: Path, glob: str) -> Generator[Path, None, None]: + path_s = str(path) + for key in self.fs.keys(): + key_s = str(key) + if key_s.startswith(path_s): + if fnmatch(key_s, glob): + yield key + + def read(self, path: Path) -> str: + return self.fs[path] + + def write(self, path: Path, content: str): + base = str(path.parent) + assert any( + [str(key).startswith(base) for key in self.fs] + ), "No parent folder, this will probably fail without a call to mkdir in a real file system!" + self.fs[path] = content + + def stat(self, path: Path): + if path in self.fs: + return Stat(path, True, True) + for item in self.fs.keys(): + if str(item).startswith(str(path)): + return Stat(path, True, False) + return Stat(path, False, False) + + def mkdir(self, path: Path): + self.fs.setdefault(path, "") + + def list(self, path: Path) -> List[Path]: + return [item for item in self.fs.keys() if item.parent == path] + + +fs = PathFs()