dreagonmon/backup_box
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

WIP VFS

Browse Source
This commit is contained in:
Dreagonmon
2025-11-04 17:39:08 +08:00
parent 60a26d9860
commit 5c564dfc66
4 changed files with 140 additions and 95 deletions
Download Patch File Download Diff File Expand all files Collapse all files

29
backup_box/app.py
View File

@@ -39,15 +39,24 @@ def main():
from .storage.vnode import VFileSystem, new_storage_file, new_storage_dir, dump_storage_item_lines from .storage.vnode import VFileSystem, new_storage_file, new_storage_dir, dump_storage_item_lines
from pathlib import PurePosixPath from pathlib import PurePosixPath
fs = VFileSystem() fs = VFileSystem()
d = new_storage_dir("dirs1", "dirt1", [ d1 = new_storage_dir("dirt1", [
new_storage_file("srn1", "tgn1"), new_storage_file("tgn1"),
new_storage_file("srn2", "tgn2"), new_storage_file("tgn2"),
new_storage_file("srn3", "tgn3"), new_storage_file("tgn3"),
new_storage_file("srn4", "tgn4"), new_storage_file("tgn4"),
]) ])
fs.root["c"].append(d) d2 = new_storage_dir("dirt2", [
fs.root["c"].append(new_storage_file(t="fsafsa")) d1,
print(fs.listdir(PurePosixPath("dirt1"))) new_storage_file("tgn1"),
for line in dump_storage_item_lines(fs.root): new_storage_file("tgn2"),
print(line) new_storage_file("tgn3"),
new_storage_file("tgn4"),
])
d3 = new_storage_dir("dirt3", [])
fs._root["c"].append(d2)
fs._root["c"].append(new_storage_file(n="fsafsa"))
fs._root["c"].append(d3)
print(fs)
fs.remove(PurePosixPath("dirt2/dirt1/tgn1"))
print(fs)

30
backup_box/storage/local_storage.py
View File

@@ -3,6 +3,7 @@ from .storage import Storage
from pathlib import PurePath, PurePosixPath from pathlib import PurePath, PurePosixPath
from ..config import StorageItem from ..config import StorageItem
from .vnode import VFileSystem from .vnode import VFileSystem
from typing import BinaryIO, Literal
class LocalStorage(Storage): class LocalStorage(Storage):
def __init__(self, base_path: str) -> None: def __init__(self, base_path: str) -> None:
@@ -16,34 +17,5 @@ class LocalStorage(Storage):
raise ValueError("config type mismatch!") raise ValueError("config type mismatch!")
return LocalStorage(cfg["path"]) 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

31
backup_box/storage/storage.py
View File

@@ -1,8 +1,8 @@
from ..config import StorageItem from ..config import StorageItem
from abc import ABC, abstractmethod from abc import ABC, abstractmethod
from pathlib import PurePosixPath, PurePath from pathlib import PurePosixPath
from collections.abc import Awaitable from collections.abc import Awaitable, Iterable, AsyncIterable
from inspect import isawaitable from typing import BinaryIO, Literal
class Storage(ABC): class Storage(ABC):
@classmethod @classmethod
@@ -14,9 +14,9 @@ class Storage(ABC):
def from_config(cls, cfg: StorageItem) -> 'Storage': def from_config(cls, cfg: StorageItem) -> 'Storage':
... ...
@classmethod
@abstractmethod @abstractmethod
def support_random_access(cls) -> bool: def sync_storage() -> Iterable[tuple[int,int]] | AsyncIterable[tuple[int,int]]:
""" sync storage, yield (current, total) progress """
... ...
@abstractmethod @abstractmethod
@@ -40,26 +40,21 @@ class Storage(ABC):
... ...
@abstractmethod @abstractmethod
def get_source_path(self, path: PurePosixPath) -> PurePath | Awaitable[PurePath]: def open(self, path: PurePosixPath, mode: Literal["r", "w"] = "r") -> BinaryIO | Awaitable[BinaryIO]:
""" get the file's source path (the location when it was backuped) """ """ open file-like object for file operation """
... ...
@abstractmethod @abstractmethod
def get_storage_path(self, path: PurePath) -> PurePosixPath | Awaitable[PurePosixPath]: def remove(self, path: PurePosixPath) -> None | Awaitable[None]:
""" get the file's storage path (the location in the storage) """ """ delete file or dir from storage, dir should be empty """
... ...
@abstractmethod @abstractmethod
def backup_file(self, source: PurePath, target: PurePosixPath) -> None | Awaitable[None]: def rmtree(self, path: PurePosixPath) -> None | Awaitable[None]:
""" copy file from local to storage """ """ remove dir tree, remove all contents """
... ...
@abstractmethod @abstractmethod
def restore_file(self, source: PurePosixPath, target: PurePath) -> None | Awaitable[None]: def makedirs(self, path: PurePosixPath, exists_ok=False) -> None | Awaitable[None]:
""" copy file from storage to local """ """ create dir """
...
@abstractmethod
def remove(self, path: PurePosixPath) -> None | Awaitable[None]:
""" delete file or dir from storage """
... ...

143
backup_box/storage/vnode.py
View File

@@ -1,27 +1,27 @@
from typing import TypedDict, Literal from typing import TypedDict, Literal, BinaryIO
from pathlib import PurePath, PurePosixPath from collections.abc import Awaitable
from pathlib import PurePosixPath
from functools import cmp_to_key from functools import cmp_to_key
from io import BytesIO
class StorageFile(TypedDict): class StorageFile(TypedDict):
ty: Literal["file"] # type ty: Literal["file"] # type
s: str # source name n: str # virtual file name
t: str # target name
mt: int # modify time mt: int # modify time
sz: int # size sz: int # size
class StorageDir(TypedDict): class StorageDir(TypedDict):
ty: Literal["dir"] # type ty: Literal["dir"] # type
s: str # source name n: str # virtual file name
t: str # target name
c: list["StorageItem"] # dir content c: list["StorageItem"] # dir content
StorageItem = StorageFile | StorageDir StorageItem = StorageFile | StorageDir
def new_storage_dir(s="", t="", c=[]) -> StorageDir: def new_storage_dir(n="", c=[]) -> StorageDir:
return StorageDir(ty="dir", s=s, t=t, c=c) return StorageDir(ty="dir", n=n, c=c)
def new_storage_file(s="", t="", mt=0, sz=0) -> StorageFile: def new_storage_file(n="", mt=0, sz=0) -> StorageFile:
return StorageFile(ty="file", s=s, t=t, mt=mt, sz=sz) return StorageFile(ty="file", n=n, mt=mt, sz=sz)
class VFSError(RuntimeError): class VFSError(RuntimeError):
def __init__(self, *args: object) -> None: def __init__(self, *args: object) -> None:
@@ -31,22 +31,26 @@ class FileNotFoundError(VFSError):
def __init__(self, p) -> None: def __init__(self, p) -> None:
super().__init__(f"FileNotFound: {p}") super().__init__(f"FileNotFound: {p}")
class FileIsNotDirError(VFSError): class NotADirectoryError(VFSError):
def __init__(self, p) -> None: def __init__(self, p) -> None:
super().__init__(f"FileIsNotDir: {p}") super().__init__(f"NotADirectory: {p}")
class DirectoryNotEmptyError(VFSError):
def __init__(self, p) -> None:
super().__init__(f"DirectoryNotEmpty: {p}")
def find_vnode(root: StorageDir, path: PurePath, search_key: Literal["s", "t"]) -> list[StorageItem]: def find_vnode(root: StorageDir, path: PurePosixPath) -> list[StorageItem]:
if len(path.parts) > 0: if len(path.parts) > 0:
p = path.parts[0] p = path.parts[0]
for n in root["c"]: for n in root["c"]:
if n[search_key] == p: if n["n"] == p:
if len(path.parts) == 1: if len(path.parts) == 1:
return [n, ] return [n, ]
if n["ty"] != "dir": if n["ty"] != "dir":
raise FileNotFoundError(p) raise FileNotFoundError(p)
next_path = PurePosixPath(*path.parts[1:]) if search_key == "t" else PurePath(*path.parts[1:]) next_path = PurePosixPath(*path.parts[1:])
nodes = find_vnode(n, next_path, search_key) nodes = find_vnode(n, next_path)
return [n, *nodes] return [n, *nodes]
raise FileNotFoundError(path) raise FileNotFoundError(path)
return [root] return [root]
@@ -59,9 +63,9 @@ def _cmp_storage_item1(item1: StorageItem, item2: StorageItem):
else: else:
return 1 return 1
# compare name # compare name
if item1["t"] > item2["t"]: if item1["n"] > item2["n"]:
return 1 return 1
elif item1["t"] < item2["t"]: elif item1["n"] < item2["n"]:
return -1 return -1
return 0 return 0
@@ -73,9 +77,9 @@ def _cmp_storage_item2(item1: StorageItem, item2: StorageItem):
else: else:
return -1 return -1
# compare name # compare name
if item1["t"] > item2["t"]: if item1["n"] > item2["n"]:
return 1 return 1
elif item1["t"] < item2["t"]: elif item1["n"] < item2["n"]:
return -1 return -1
return 0 return 0
@@ -85,9 +89,9 @@ def sort_storage_items(items: list[StorageItem], dir_first = True):
def dump_storage_item_lines(item: StorageItem, indent=0): def dump_storage_item_lines(item: StorageItem, indent=0):
indent_text = " " * indent indent_text = " " * indent
if item["ty"] == "file": if item["ty"] == "file":
return [f"{indent_text}📄{item["t"]}: size {item["sz"]}"] return [f"{indent_text}📄{item["n"]}: size {item["sz"]}"]
elif item["ty"] == "dir": elif item["ty"] == "dir":
lines = [f"{indent_text}📂{item["t"]}:"] lines = [f"{indent_text}📂{item["n"]}:"]
sort_storage_items(item["c"], False) sort_storage_items(item["c"], False)
for n in item["c"]: for n in item["c"]:
lines.extend(dump_storage_item_lines(n, indent=indent + 1)) lines.extend(dump_storage_item_lines(n, indent=indent + 1))
@@ -96,35 +100,100 @@ def dump_storage_item_lines(item: StorageItem, indent=0):
class VFileSystem(): class VFileSystem():
def __init__(self) -> None: def __init__(self) -> None:
self.root = new_storage_dir() self._root = new_storage_dir()
def __repr__(self):
return "\n".join(dump_storage_item_lines(self._root))
def _on_remove_node(self, node_path: list[StorageItem]) -> None | Awaitable[None]:
parents_path = PurePosixPath( *(n["n"] for n in node_path) )
print("remove:", parents_path)
pass
def _op_open_node(self, node_path: list[StorageItem]) -> BinaryIO | Awaitable[BinaryIO]:
return BytesIO()
def listdir(self, path=PurePosixPath("")) -> list[PurePosixPath]: def listdir(self, path=PurePosixPath("")) -> list[PurePosixPath]:
nodes = find_vnode(self.root, path, "t") nodes = find_vnode(self._root, path)
base_path = PurePosixPath( *(n["t"] for n in nodes) ) base_path = PurePosixPath( *(n["n"] for n in nodes) )
item = nodes[-1] item = nodes[-1]
if item["ty"] != "dir": if item["ty"] != "dir":
raise FileIsNotDirError(path) raise NotADirectoryError(path)
sort_storage_items(item["c"], True) sort_storage_items(item["c"], True)
return [ base_path.joinpath(n["t"]) for n in item["c"]] return [ base_path.joinpath(n["n"]) for n in item["c"]]
def exists(self, path: PurePosixPath) -> bool: def exists(self, path: PurePosixPath) -> bool:
try: try:
_ = find_vnode(self.root, path, "t") _ = find_vnode(self._root, path)
return True return True
except FileNotFoundError: except FileNotFoundError:
return False return False
def is_dir(self, path: PurePosixPath) -> bool: def is_dir(self, path: PurePosixPath) -> bool:
raise NotImplementedError node = find_vnode(self._root, path)[-1]
return node["ty"] == "dir"
def is_file(self, path: PurePosixPath) -> bool: def is_file(self, path: PurePosixPath) -> bool:
raise NotImplementedError node = find_vnode(self._root, path)[-1]
return node["ty"] == "file"
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: def remove(self, path: PurePosixPath) -> None:
raise NotImplementedError nodes = find_vnode(self._root, path)
if len(nodes) == 1:
# in the root
parent = self._root
elif len(nodes) >= 2:
parent = nodes[-2]
else:
raise FileNotFoundError(path)
last = nodes[-1]
if last["ty"] == "dir":
if len(last["c"]) > 0:
raise DirectoryNotEmptyError(path)
# remove callback
self._on_remove_node(nodes)
if parent["ty"] == "dir":
parent["c"].remove(last)
def rmtree(self, path: PurePosixPath) -> None:
nodes = find_vnode(self._root, path)
if len(nodes) == 1:
# in the root
parent = self._root
prefix = []
elif len(nodes) >= 2:
parent = nodes[-2]
prefix = nodes[:-1]
else:
raise FileNotFoundError(path)
last = nodes[-1]
vstack = [last]
while True:
last = vstack[-1]
# process dir
if last["ty"] == "dir":
if len(last["c"]) > 0:
# remove child at the last loop
vstack.append(last["c"][0])
parent = last
continue
# remove callback
node_path = list(prefix)
node_path.extend(vstack)
self._on_remove_node(node_path)
# remove self from parent
if parent["ty"] == "dir":
parent["c"].remove(last)
vstack.pop()
if len(vstack) >= 2:
parent = vstack[-2]
elif len(vstack) == 1:
parent = prefix[-1] if len(prefix) >= 1 else self._root
else:
pass
else:
raise NotADirectoryError(parent["n"])
# check if anything left
if len(vstack) <= 0:
break