diff --git a/backup_box/app.py b/backup_box/app.py index 7d1382e..b7510b9 100644 --- a/backup_box/app.py +++ b/backup_box/app.py @@ -36,5 +36,18 @@ def main(): parse_args() print("Config:", config.get_config()) print("Hello World") - from .storage import LocalStorage - print(LocalStorage.get_storage_type()) + from .storage.vnode import VFileSystem, new_storage_file, new_storage_dir, dump_storage_item_lines + from pathlib import PurePosixPath + fs = VFileSystem() + d = new_storage_dir("dirs1", "dirt1", [ + new_storage_file("srn1", "tgn1"), + new_storage_file("srn2", "tgn2"), + new_storage_file("srn3", "tgn3"), + new_storage_file("srn4", "tgn4"), + ]) + fs.root["c"].append(d) + fs.root["c"].append(new_storage_file(t="fsafsa")) + print(fs.listdir(PurePosixPath("dirt1"))) + for line in dump_storage_item_lines(fs.root): + print(line) + diff --git a/backup_box/config.py b/backup_box/config.py index 4687b59..3d14142 100644 --- a/backup_box/config.py +++ b/backup_box/config.py @@ -6,6 +6,7 @@ from typing import TypedDict, Literal EntryItem = TypedDict("EntryItem", { "source": str, + "included": list[str], "ignored": list[str], }) @@ -70,6 +71,7 @@ def apply_user_config(dir_or_path: str): # path related to the config file. if item["source"]: item["source"] = _path_related_to(cfg_dir, item["source"]) + item.setdefault("included", []) item.setdefault("ignored", []) config.setdefault("storage", {}) for item in config["storage"].values(): diff --git a/backup_box/storage/local_storage.py b/backup_box/storage/local_storage.py index 9521c87..cda3ac5 100644 --- a/backup_box/storage/local_storage.py +++ b/backup_box/storage/local_storage.py @@ -1,4 +1,49 @@ +from collections.abc import Awaitable from .storage import Storage +from pathlib import PurePath, PurePosixPath +from ..config import StorageItem +from .vnode import VFileSystem class LocalStorage(Storage): - pass + def __init__(self, base_path: str) -> None: + super().__init__() + self.base = PurePath(base_path) + self.vfs = VFileSystem() + + @classmethod + def from_config(cls, cfg: StorageItem) -> "LocalStorage": + if cfg["type"] != "LocalStorageItem": + raise ValueError("config type mismatch!") + return LocalStorage(cfg["path"]) + + @classmethod + def support_random_access(cls) -> bool: + return True + + def listdir(self, path=PurePosixPath("")) -> list[PurePosixPath] | Awaitable[list[PurePosixPath]]: + raise NotImplementedError + + def exists(self, path: PurePosixPath) -> bool | Awaitable[bool]: + raise NotImplementedError + + def is_dir(self, path: PurePosixPath) -> bool | Awaitable[bool]: + raise NotImplementedError + + def is_file(self, path: PurePosixPath) -> bool | Awaitable[bool]: + raise NotImplementedError + + def get_source_path(self, path: PurePosixPath) -> PurePath | Awaitable[PurePath]: + raise NotImplementedError + + def get_storage_path(self, path: PurePath) -> PurePosixPath | Awaitable[PurePosixPath]: + raise NotImplementedError + + def backup_file(self, source: PurePath, target: PurePosixPath) -> None | Awaitable[None]: + raise NotImplementedError + + def restore_file(self, source: PurePosixPath, target: PurePath) -> None | Awaitable[None]: + raise NotImplementedError + + def remove(self, path: PurePosixPath) -> None | Awaitable[None]: + raise NotImplementedError + diff --git a/backup_box/storage/storage.py b/backup_box/storage/storage.py index c47f064..346c219 100644 --- a/backup_box/storage/storage.py +++ b/backup_box/storage/storage.py @@ -1,5 +1,8 @@ -from abc import ABC, abstractmethod from ..config import StorageItem +from abc import ABC, abstractmethod +from pathlib import PurePosixPath, PurePath +from collections.abc import Awaitable +from inspect import isawaitable class Storage(ABC): @classmethod @@ -10,3 +13,53 @@ class Storage(ABC): @abstractmethod def from_config(cls, cfg: StorageItem) -> 'Storage': ... + + @classmethod + @abstractmethod + def support_random_access(cls) -> bool: + ... + + @abstractmethod + def listdir(self, path=PurePosixPath("")) -> list[PurePosixPath] | Awaitable[list[PurePosixPath]]: + """ list storage file in 'path' """ + ... + + @abstractmethod + def exists(self, path: PurePosixPath) -> bool | Awaitable[bool]: + """ if the file or dir is exists """ + ... + + @abstractmethod + def is_dir(self, path: PurePosixPath) -> bool | Awaitable[bool]: + """ if the path is dir """ + ... + + @abstractmethod + def is_file(self, path: PurePosixPath) -> bool | Awaitable[bool]: + """ if the path is file """ + ... + + @abstractmethod + def get_source_path(self, path: PurePosixPath) -> PurePath | Awaitable[PurePath]: + """ get the file's source path (the location when it was backuped) """ + ... + + @abstractmethod + def get_storage_path(self, path: PurePath) -> PurePosixPath | Awaitable[PurePosixPath]: + """ get the file's storage path (the location in the storage) """ + ... + + @abstractmethod + def backup_file(self, source: PurePath, target: PurePosixPath) -> None | Awaitable[None]: + """ copy file from local to storage """ + ... + + @abstractmethod + def restore_file(self, source: PurePosixPath, target: PurePath) -> None | Awaitable[None]: + """ copy file from storage to local """ + ... + + @abstractmethod + def remove(self, path: PurePosixPath) -> None | Awaitable[None]: + """ delete file or dir from storage """ + ... diff --git a/backup_box/storage/vnode.py b/backup_box/storage/vnode.py new file mode 100644 index 0000000..bcd6bb0 --- /dev/null +++ b/backup_box/storage/vnode.py @@ -0,0 +1,130 @@ +from typing import TypedDict, Literal +from pathlib import PurePath, PurePosixPath +from functools import cmp_to_key + +class StorageFile(TypedDict): + ty: Literal["file"] # type + s: str # source name + t: str # target name + mt: int # modify time + sz: int # size + +class StorageDir(TypedDict): + ty: Literal["dir"] # type + s: str # source name + t: str # target name + c: list["StorageItem"] # dir content + +StorageItem = StorageFile | StorageDir + +def new_storage_dir(s="", t="", c=[]) -> StorageDir: + return StorageDir(ty="dir", s=s, t=t, c=c) + +def new_storage_file(s="", t="", mt=0, sz=0) -> StorageFile: + return StorageFile(ty="file", s=s, t=t, mt=mt, sz=sz) + +class VFSError(RuntimeError): + def __init__(self, *args: object) -> None: + super().__init__(*args) + +class FileNotFoundError(VFSError): + def __init__(self, p) -> None: + super().__init__(f"FileNotFound: {p}") + +class FileIsNotDirError(VFSError): + def __init__(self, p) -> None: + super().__init__(f"FileIsNotDir: {p}") + + +def find_vnode(root: StorageDir, path: PurePath, search_key: Literal["s", "t"]) -> list[StorageItem]: + if len(path.parts) > 0: + p = path.parts[0] + for n in root["c"]: + if n[search_key] == p: + if len(path.parts) == 1: + return [n, ] + if n["ty"] != "dir": + raise FileNotFoundError(p) + next_path = PurePosixPath(*path.parts[1:]) if search_key == "t" else PurePath(*path.parts[1:]) + nodes = find_vnode(n, next_path, search_key) + return [n, *nodes] + raise FileNotFoundError(path) + return [root] + +def _cmp_storage_item1(item1: StorageItem, item2: StorageItem): + """ cmp, sort dir first, then name """ + if item1["ty"] != item2["ty"]: + if item1["ty"] == "dir": + return -1 # dir in front + else: + return 1 + # compare name + if item1["t"] > item2["t"]: + return 1 + elif item1["t"] < item2["t"]: + return -1 + return 0 + +def _cmp_storage_item2(item1: StorageItem, item2: StorageItem): + """ cmp, sort file first, then name """ + if item1["ty"] != item2["ty"]: + if item1["ty"] == "dir": + return 1 # dir behind + else: + return -1 + # compare name + if item1["t"] > item2["t"]: + return 1 + elif item1["t"] < item2["t"]: + return -1 + return 0 + +def sort_storage_items(items: list[StorageItem], dir_first = True): + items.sort(key=cmp_to_key(_cmp_storage_item1 if dir_first else _cmp_storage_item2)) + +def dump_storage_item_lines(item: StorageItem, indent=0): + indent_text = " " * indent + if item["ty"] == "file": + return [f"{indent_text}📄{item["t"]}: size {item["sz"]}"] + elif item["ty"] == "dir": + lines = [f"{indent_text}📂{item["t"]}:"] + sort_storage_items(item["c"], False) + for n in item["c"]: + lines.extend(dump_storage_item_lines(n, indent=indent + 1)) + return lines + return [] + +class VFileSystem(): + def __init__(self) -> None: + self.root = new_storage_dir() + + def listdir(self, path=PurePosixPath("")) -> list[PurePosixPath]: + nodes = find_vnode(self.root, path, "t") + base_path = PurePosixPath( *(n["t"] for n in nodes) ) + item = nodes[-1] + if item["ty"] != "dir": + raise FileIsNotDirError(path) + sort_storage_items(item["c"], True) + return [ base_path.joinpath(n["t"]) for n in item["c"]] + + def exists(self, path: PurePosixPath) -> bool: + try: + _ = find_vnode(self.root, path, "t") + return True + except FileNotFoundError: + return False + + def is_dir(self, path: PurePosixPath) -> bool: + raise NotImplementedError + + def is_file(self, path: PurePosixPath) -> bool: + raise NotImplementedError + + def get_source_path(self, path: PurePosixPath) -> PurePath: + raise NotImplementedError + + def get_storage_path(self, path: PurePath) -> PurePosixPath: + raise NotImplementedError + + def remove(self, path: PurePosixPath) -> None: + raise NotImplementedError