WIP VFileSystem

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)
+

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():

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
+

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 """
+        ...

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