Centralised FileManager core functionalities

Author: Dtar380

src/cli/builder.py

@@ -2,18 +2,13 @@
 
 import inspect
 from pathlib import Path
-from typing import Any, cast
+from typing import Any
 
 from InquirerPy import inquirer  # type: ignore
 from InquirerPy.validator import EmptyInputValidator  # type: ignore
 from click import Command, Option
-from importlib_resources import as_file, files  # type: ignore
-from yaspin import yaspin  # type: ignore
 
-from ..core.copy_files import copy_files  # type: ignore
-from ..core.docker import ComposeManager
-from ..core.manage_json import read_json, write_json
-from ..core.manage_templates import template_to_file
+from ..core.files import FileManager
 from ..utils.cli import clear, confirm
 from .custom_group import CustomGroup
 from .menu import Menus
@@ -25,6 +20,7 @@ class Builder(CustomGroup):
 
     def __init__(self) -> None:
         super().__init__()
+        self.file_manager = FileManager()
 
     def create(self) -> Command:
         help = (
@@ -68,7 +64,7 @@ def callback(network: bool = False) -> None:
                         break
 
             clear(0)
-            self.__save_files(
+            self.file_manager.save_files(
                 data={
                     "compose": {
                         "services": services,
@@ -107,7 +103,7 @@ def callback(
                     "ERROR: Missing JSON file for services. Use 'create' first."
                 )
 
-            data: dicts = read_json(path) or {}
+            data: dicts = self.file_manager.read_json(path) or {}
             compose: dicts = data.get("compose", {}) or {}
 
             services: list[dicts] = compose.get("services", []) or []
@@ -147,7 +143,7 @@ def find_index_by_name(name: str) -> int | None:
                     compose["networks"] = networks
                     data["compose"] = compose
                     data["envs"] = envs
-                    self.__save_files(data)
+                    self.file_manager.save_files(data)
                     print(f"Service '{target}' removed and files updated.")
 
             elif add:
@@ -182,7 +178,7 @@ def find_index_by_name(name: str) -> int | None:
                     compose["networks"] = networks
                     data["compose"] = compose
                     data["envs"] = envs
-                    self.__save_files(data)
+                    self.file_manager.save_files(data)
                     print(f"Service '{name}' removed and files updated.")
 
             else:
@@ -212,13 +208,13 @@ def callback() -> None:
                     "ERROR: Missing JSON file for services. Use 'create' first."
                 )
 
-            data: dicts = read_json(path) or {}
+            data: dicts = self.file_manager.read_json(path) or {}
 
             if not data:
                 exit("ERROR: JSON file is empty. Use 'create' first.")
 
             clear(0)
-            self.__save_files(data, build=True)
+            self.file_manager.save_files(data, build=True)
             clear(0)
             print("Files saved!")
 
@@ -253,32 +249,3 @@ def __get_name(self, message: str) -> str:
                 break
 
         return name
-
-    @yaspin(text="Saving files...", color="cyan")
-    def __save_files(self, data: dicts, build: bool = False) -> None:
-        tmps_path = files("minecraft-docker-cli.assets.templates")
-        composer_template = tmps_path.joinpath("docker-compose.yml.j2")
-        env_template = tmps_path.joinpath(".env.j2")
-
-        if not build:
-            write_json(self.cwd.joinpath("data.json"), data)
-
-        composer: dicts = data.get("composer") or {}
-        with as_file(composer_template) as composer_path:  # type: ignore
-            composer_path = cast(Path, composer_path)
-            template_to_file(
-                composer_path, composer, self.cwd.joinpath("docker-compose.yml")
-            )
-
-        services: list[dicts] = composer.get("services", []) or []
-        names: list[str] = [service.get("name") for service in services]  # type: ignore
-        copy_files(self.cwd, names)
-
-        envs: list[dicts] = data.get("envs") or []
-        for env in envs:
-            relative_path = f"servers/{env.get("CONTAINER_NAME")}/.env"  # type: ignore
-            with as_file(env_template) as env_path:  # type: ignore
-                env_path = cast(Path, env_path)
-                template_to_file(
-                    env_path, env, self.cwd.joinpath(relative_path)
-                )

src/cli/manager.py

@@ -7,7 +7,6 @@
 from click import Command, Option
 
 from ..core.docker import ComposeManager
-from ..utils.cli import clear, confirm  # type: ignore
 from .custom_group import CustomGroup
 
 

src/core/copy_files.py

@@ -7,7 +7,7 @@
 
 from yaspin import yaspin
 
-from .manage_json import read_json
+from .files import FileManager
 
 
 class ComposeManager:
@@ -18,6 +18,7 @@ def __init__(self) -> None:
             exit(
                 "ERROR: docker-compose.yml was not located in current directory."
             )
+        self.file_manager = FileManager()
 
     def __run(
         self, *args: str, capture_output: bool = False
@@ -30,7 +31,7 @@ def __run(
             print("Command run:\n", result.stdout)
         return result
 
-    @yaspin("Building Container...")
+    @yaspin("Building Container...", color="cyan")
     def build(
         self, no_cache: bool = False, pull: bool = False
     ) -> CompletedProcess[str]:
@@ -41,35 +42,35 @@ def build(
             args.append("--pull")
         return self.__run(*args)
 
-    @yaspin("Stopping Services...")
+    @yaspin("Stopping Services...", color="cyan")
     def stop(self) -> CompletedProcess[str]:
         return self.__run("stop")
 
-    @yaspin("Starting Services...")
+    @yaspin("Starting Services...", color="cyan")
     def start(self) -> CompletedProcess[str]:
         return self.__run("start")
 
-    @yaspin("Removing Container...")
+    @yaspin("Removing Container...", color="cyan")
     def down(self, remove_volumes: bool = False) -> CompletedProcess[str]:
         args = ["down"]
         if remove_volumes:
             args.append("-v")
         return self.__run(*args)
 
-    @yaspin("Putting Up Container...")
+    @yaspin("Putting Up Container...", color="cyan")
     def up(self, detached: bool = True) -> CompletedProcess[str]:
         args = ["up"]
         if detached:
             args.append("-d")
         return self.__run(*args)
 
-    @yaspin("Backing Up Container...")
+    @yaspin("Backing Up Container...", color="cyan")
     def back_up(self, cwd: Path = Path.cwd()) -> None:
         backup_path = cwd.joinpath(".backup")
         compose_json = cwd.joinpath("data.json")
 
         backup_path.mkdir(exist_ok=True)
-        data: dict[str, Any] = read_json(compose_json)
+        data: dict[str, Any] = self.file_manager.read_json(compose_json)
         services = data.get("composer", {}).get("services", []) or []
         names: list[str] = [svc.get("name") for svc in services if svc.get("name") is not None]  # type: ignore
         for svc_name in names:

src/core/docker.py

@@ -0,0 +1,107 @@
+from __future__ import annotations
+
+import json
+from pathlib import Path
+from typing import Any, cast
+
+from importlib_resources import as_file, files  # type: ignore
+import jinja2
+from yaspin import yaspin  # type: ignore
+
+dicts = dict[str, Any]
+
+
+class FileManager:
+
+    cwd = Path.cwd()
+
+    def save_files(self, data: dicts, build: bool = False) -> None:
+        tmps_path = files("minecraft-docker-cli.assets.templates")
+        composer_template = tmps_path.joinpath("docker-compose.yml.j2")
+        env_template = tmps_path.joinpath(".env.j2")
+
+        if not build:
+            self.write_json(self.cwd.joinpath("data.json"), data)
+
+        composer: dicts = data.get("composer") or {}
+        with as_file(composer_template) as composer_path:  # type: ignore
+            composer_path = cast(Path, composer_path)
+            self.template_to_file(
+                composer_path, composer, self.cwd.joinpath("docker-compose.yml")
+            )
+
+        services: list[dicts] = composer.get("services", []) or []
+        names: list[str] = [service.get("name") for service in services]  # type: ignore
+        self.copy_files(self.cwd, names)
+
+        envs: list[dicts] = data.get("envs") or []
+        for env in envs:
+            relative_path = f"servers/{env.get("CONTAINER_NAME")}/.env"  # type: ignore
+            with as_file(env_template) as env_path:  # type: ignore
+                env_path = cast(Path, env_path)
+                self.template_to_file(
+                    env_path, env, self.cwd.joinpath(relative_path)
+                )
+
+    @yaspin("Reading JSON...", color="cyan")
+    def read_json(self, file: Path) -> dict[Any, Any]:
+        with open(file, "r+") as f:
+            data = dict(json.load(f))
+        return data
+
+    @yaspin("Writting JSON...", color="cyan")
+    def write_json(self, file: Path, data: dict[Any, Any]) -> None:
+        data_str = json.dumps(data, indent=2)
+        with open(file, "w+") as f:
+            f.write(data_str)
+        return None
+
+    @yaspin("Copying files...", color="cyan")
+    def copy_files(self, path: Path, services: list[str]) -> None:
+        docker_pkg = files("minecraft-docker-cli.assets.docker")
+        dockerfile_res = docker_pkg.joinpath("Dockerfile")
+        dockerignore_res = docker_pkg.joinpath(".dockerignore")
+        runsh_res = files("minecraft-docker-cli.assets.scripts").joinpath(
+            "run.sh"
+        )
+        readme_res = files("minecraft-docker-cli.assets").joinpath("README.md")
+
+        # Ensure base path exists
+        if not path.exists():
+            raise ValueError("Path doesnt exist")
+
+        # Read bytes from resources once
+        dockerfile_bytes = dockerfile_res.read_bytes()
+        dockerignore_bytes = dockerignore_res.read_bytes()
+        runsh_bytes = runsh_res.read_bytes()
+        readme_bytes = readme_res.read_bytes()
+
+        # Write files for each service
+        for service in services:
+            dest_dir = path.joinpath("servers", service)
+            dest_dir.mkdir(parents=True, exist_ok=True)
+
+            (dest_dir / "Dockerfile").write_bytes(dockerfile_bytes)
+            (dest_dir / ".dockerignore").write_bytes(dockerignore_bytes)
+            (dest_dir / "run.sh").write_bytes(runsh_bytes)
+
+        # Write top-level README into the given path
+        (path / "README.md").write_bytes(readme_bytes)
+
+    @yaspin("Rendering template...", color="cyan")
+    def template_to_file(
+        self, template_path: Path, context: dict[Any, Any], dest_path: Path
+    ) -> Path:
+        rendered = self.__render_template(template_path, context)
+        dest_path.parent.mkdir(parents=True, exist_ok=True)
+        dest_path.write_text(rendered, encoding="utf-8")
+        return dest_path
+
+    def __render_template(
+        self, template_path: Path, context: dict[Any, Any]
+    ) -> str:
+        env = jinja2.Environment(
+            loader=jinja2.FileSystemLoader(str(template_path.parent))
+        )
+        template_obj = env.get_template(template_path.name)
+        return template_obj.render(**context)