add file size and modify time

backup_box/app.py

@@ -46,6 +46,9 @@ async def a_main():
     fs = LocalStorage.from_config(stm)
     async for rt in fs.sync_storage():
         print(rt)
+    async with FCTX(fs.open(PurePosixPath("app.txt"), "w")) as fp:
+        await fp.write(b"Hello Wyvern!")
+    await fs.set_last_modify_time(PurePosixPath("app.txt"), 0)
     print(fs.dump_fs_tree())
 
 def main():

backup_box/storage/local_storage.py

@@ -1,16 +1,22 @@
 from backup_box.storage.storage import AsyncFileReader, AsyncFileWriter
-from pathlib import PurePosixPath, PurePath
+from pathlib import PurePath
 from ..config import StorageItem
 from .vnode import VFileSystem, VFile, VDir, new_storage_dir, new_storage_file
 from typing import Literal, overload
 from stat import S_ISDIR
 from aiofiles import open
 from aiofiles.os import remove, rmdir, stat, listdir, mkdir
+from concurrent.futures import ThreadPoolExecutor
+from asyncio import get_event_loop
+from os import utime as utime_sync
+from functools import partial
+from time import time
 
 class LocalStorage(VFileSystem):
     def __init__(self, base_path: str) -> None:
         super().__init__()
         self._base = PurePath(base_path)
+        self._exec = ThreadPoolExecutor(1)
     
     @classmethod
     def from_config(cls, cfg: StorageItem) -> "LocalStorage":
@@ -60,13 +66,22 @@ class LocalStorage(VFileSystem):
     async def _on_create_node(self, node_path: list[VFile | VDir]) -> None:
         if len(node_path) > 0 and node_path[-1]["ty"] == "dir":
             real_path = self._vnode_path_to_syspath(*node_path)
+            print(real_path)
             await mkdir(real_path)
 
+    async def _on_set_last_modify_time(self, node_path: list[VFile | VDir]) -> None:
+        if len(node_path) <= 0 or node_path[-1]["ty"] != "file":
+            raise IsADirectoryError()
+        real_path = self._vnode_path_to_syspath(*node_path)
+        await get_event_loop().run_in_executor(self._exec, utime_sync, real_path, (int(time()), node_path[-1]["mt"]))
+
     @overload
     async def _on_open_node(self, node_path: list[VFile | VDir], mode: Literal["r"] = "r") -> AsyncFileReader: ...
     @overload
     async def _on_open_node(self, node_path: list[VFile | VDir], mode: Literal["w"]) -> AsyncFileWriter: ...
     async def _on_open_node(self, node_path: list[VFile | VDir], mode: Literal["r"] | Literal["w"] = "r") -> AsyncFileWriter | AsyncFileReader:
+        if len(node_path) <= 0:
+            raise IsADirectoryError()
         real_path = self._vnode_path_to_syspath(*node_path)
         if mode == "r":
             return await open(real_path, "rb")
@@ -74,8 +89,3 @@ class LocalStorage(VFileSystem):
             return await open(real_path, "wb")
         else:
             raise ValueError()
-        
-
-
-
-

backup_box/storage/storage.py

@@ -20,15 +20,17 @@ class AsyncFileWriter(Protocol):
         ...
 
 class AsyncFileContextManager[T](AbstractAsyncContextManager[T]):
-    def __init__(self, fobj: T) -> None:
+    def __init__(self, fobj: Awaitable[T]) -> None:
         self.__fobj = fobj
+        self.__obj: T | None = None
 
     async def __aenter__(self) -> T:
-        return self.__fobj # type: ignore
+        self.__obj = await self.__fobj
+        return self.__obj # type: ignore
 
     async def __aexit__(self, exc_type, exc, tb):
-        if hasattr(self.__fobj, "close"):
-            func = getattr(self.__fobj, "close")
+        if hasattr(self.__obj, "close"):
+            func = getattr(self.__obj, "close")
             await func()
 
 class Storage(ABC):
@@ -91,3 +93,18 @@ class Storage(ABC):
     def makedirs(self, path: PurePosixPath, exists_ok=False) -> Awaitable[None]:
         """ create dir """
         ...
+
+    @abstractmethod
+    def get_file_size(self, path: PurePosixPath) -> Awaitable[int]:
+        """ get file size, can raise error for dir """
+        ...
+
+    @abstractmethod
+    def get_last_modify_time(self, path: PurePosixPath) -> Awaitable[int]:
+        """ get last modify time in seconds since unix time, can raise error for dir """
+        ...
+    
+    @abstractmethod
+    def set_last_modify_time(self, path: PurePosixPath, timestamp: int) -> Awaitable[None]:
+        """ get last modify time in seconds since unix time, can raise error for dir """
+        ...

backup_box/storage/vnode.py

@@ -7,7 +7,7 @@ from abc import abstractmethod
 class VFile(TypedDict):
     ty: Literal["file"] # type
     n: str # virtual file name
-    mt: int # modify time
+    mt: int # modify time seconds
     sz: int # size
 
 class VDir(TypedDict):
@@ -102,6 +102,13 @@ class VFileSystem(Storage):
         path = PurePosixPath( *(n["n"] for n in node_path) )
         print("create:", path)
         ...
+    
+    @abstractmethod
+    async def _on_set_last_modify_time(self, node_path: list[VFSItem]) -> None:
+        """ create a file or dir, parent dir is always created first. """
+        path = PurePosixPath( *(n["n"] for n in node_path) )
+        print("create:", path)
+        ...
 
     @overload
     @abstractmethod
@@ -240,6 +247,8 @@ class VFileSystem(Storage):
                 build.append(node)
                 await self._on_create_node(build)
         # create dir
+        if parent == self._root:
+            return
         for n in parent["c"]:
             if n["n"] == path.name:
                 if not exists_ok:
@@ -253,3 +262,25 @@ class VFileSystem(Storage):
             build.append(node)
             await self._on_create_node(build)
 
+    async def get_file_size(self, path: PurePosixPath) -> int:
+        node = find_vnode(self._root, path)[-1]
+        if node["ty"] == "file":
+            return node["sz"]
+        raise IsADirectoryError(path)
+
+    async def get_last_modify_time(self, path: PurePosixPath) -> int:
+        node = find_vnode(self._root, path)[-1]
+        if node["ty"] == "file":
+            return node["mt"]
+        else:
+            raise IsADirectoryError(path)
+    
+    async def set_last_modify_time(self, path: PurePosixPath, timestamp: int):
+        """ get last modify time in seconds since unix time """
+        nodes = find_vnode(self._root, path)
+        node = nodes[-1]
+        if node["ty"] == "file":
+            node["mt"] = timestamp
+            await self._on_set_last_modify_time(nodes)
+        else:
+            raise IsADirectoryError(path)