refactor: all file handling in new class

src/data_dir.py

@@ -0,0 +1,52 @@
+import os
+import io
+import tarfile
+import shutil
+
+
+class DataDir:
+    HOST_FILE = ".host"
+
+    def __init__(self, root_path: str):
+        self.root_path = root_path
+
+    def list_paths(self) -> list[str]:
+        paths: list[str] = []
+        for path in os.listdir(self.root_path):
+            if os.path.isdir(os.path.join(self.root_path, path)):
+                paths += [path]
+        return paths
+
+    def set_host(self, path: str, host: str):
+        path_host = os.path.join(self.root_path, path, self.HOST_FILE)
+        with open(path_host, mode="w") as host_file:
+            host_file.write(host)
+
+    def has_index(self, path: str):
+        path_index = os.path.join(self.root_path, path, "index.html")
+        return os.path.exists(path_index) and os.path.isfile(path_index)
+
+    def get_host(self, path: str):
+        path_host = os.path.join(self.root_path, path, self.HOST_FILE)
+        if os.path.exists(path_host) and os.path.isfile(path_host):
+            try:
+                with open(path_host) as host_file:
+                    return host_file.read().split("\n")[0].strip()
+            except Exception:
+                pass
+        return None
+
+    def extract_tar_bytes(self, path: str, tar_bytes: io.BytesIO):
+        target_path = os.path.join(self.root_path, path)
+        with tarfile.open(fileobj=tar_bytes) as tar_file:
+            if os.path.exists(target_path):
+                shutil.rmtree(target_path)
+            tar_file.extractall(target_path)
+
+    def remove(self, path: str):
+        target_path = os.path.join(self.root_path, path)
+        shutil.rmtree(target_path)
+
+    def exists(self, path: str):
+        target_path = os.path.join(self.root_path, path)
+        return os.path.exists(target_path) and os.path.isdir(target_path)

src/handler.py

@@ -4,9 +4,8 @@ import tarfile
 import re
 import io
 import os
-import shutil
 
-from . import project, params, registry
+from . import project, params, registry, data_dir
 
 
 class StaplerRequestHandler(http.server.SimpleHTTPRequestHandler):
@@ -18,7 +17,7 @@ class StaplerRequestHandler(http.server.SimpleHTTPRequestHandler):
     ):
         self.default_host = params.host
         self.token = params.token
-        self.data_dir = params.data_dir
+        self.data_dir = data_dir.DataDir(params.data_dir)
         self.max_size_bytes = params.max_size_bytes
         self.registry = registry
         super().__init__(*args, directory=params.data_dir, **kwargs)
@@ -54,11 +53,7 @@ class StaplerRequestHandler(http.server.SimpleHTTPRequestHandler):
             )
         try:
             file_bytes = io.BytesIO(self.rfile.read(content_length))
-            target_path = os.path.join(self.data_dir, sub_path)
-            with tarfile.open(fileobj=file_bytes) as tar_file:
-                if os.path.exists(target_path):
-                    shutil.rmtree(target_path)
-                tar_file.extractall(os.path.join(self.data_dir, sub_path))
+            self.data_dir.extract_tar_bytes(sub_path, file_bytes)
         except tarfile.TarError:
             return self.send_error(http.HTTPStatus.BAD_REQUEST, "Invalid tar archive")
         except Exception as e:
@@ -73,11 +68,10 @@ class StaplerRequestHandler(http.server.SimpleHTTPRequestHandler):
             return self.send_error(http.HTTPStatus.UNAUTHORIZED, "Invalid token")
         if (sub_path := self.get_subpath()) is None:
             return self.send_error(http.HTTPStatus.BAD_REQUEST, "Invalid path")
-        target_path = os.path.join(self.data_dir, sub_path)
-        if not os.path.exists(target_path):
+        if not self.data_dir.exists(sub_path):
             return self.send_error(http.HTTPStatus.NOT_FOUND, "Not found")
         try:
-            shutil.rmtree(target_path)
+            self.data_dir.remove(sub_path)
         except Exception as e:
             return self.send_error(http.HTTPStatus.INTERNAL_SERVER_ERROR, str(e))
         self.send_status_only(

src/registry.py

@@ -1,49 +1,27 @@
-import os
-
-from . import params, page
-
-_HOST_FILE = ".host"
+from . import params, page, data_dir
 
 
 class Registry:
     def __init__(self, params: params.Parameters):
         self.pages: dict[str, page.Page] = {}
-        self.data_dir = params.data_dir
+        self.data_dir = data_dir.DataDir(params.data_dir)
         self.prefix = f"http://{params.host}:{params.port}"
 
     def load_pages(self):
         self.pages = {}
-        for path in os.listdir(self.data_dir):
+        for path in self.data_dir.list_paths():
             self.add(path)
 
     def add(self, path: str):
-        real_path = os.path.join(self.data_dir, path)
-        if os.path.isdir(real_path):
-            self.pages[path] = page.Page(
-                path, self.__has_index(path), self.__get_host(path)
-            )
-            print("Updated: " + self.prefix + str(self.pages[path]))
+        self.pages[path] = page.Page(
+            path, self.data_dir.has_index(path), self.data_dir.get_host(path)
+        )
+        print("Updated: " + self.prefix + str(self.pages[path]))
 
     def set_host(self, path: str, host: str):
-        path_host = os.path.join(self.data_dir, path, _HOST_FILE)
-        with open(path_host, mode="w") as host_file:
-            host_file.write(host)
+        self.data_dir.set_host(path, host)
         self.pages[path].host = host
 
-    def __has_index(self, path: str) -> bool:
-        path_index = os.path.join(self.data_dir, path, "index.html")
-        return os.path.exists(path_index) and os.path.isfile(path_index)
-
-    def __get_host(self, path: str) -> str | None:
-        path_host = os.path.join(self.data_dir, path, _HOST_FILE)
-        if os.path.exists(path_host) and os.path.isfile(path_host):
-            try:
-                with open(path_host) as host_file:
-                    return host_file.read().split("\n")[0].strip()
-            except Exception:
-                pass
-        return None
-
     def remove(self, path: str):
         page = self.pages[path]
         del self.pages[path]