feat: add docker preset

project_manager/common/command.py

@@ -0,0 +1,68 @@
+__all__ = (
+    "CommandError",
+    "CommandLogger",
+    "CommandExecutor",
+)
+
+
+import logging
+import os
+import subprocess
+import threading
+from pathlib import Path
+
+logger = logging.getLogger("command")
+
+
+class CommandError(RuntimeError):
+    @property
+    def output(self):
+        return self.args[0]
+
+
+class CommandLogger(threading.Thread):
+    def __init__(self, level: int):
+        super().__init__()
+        self.daemon = False
+        self.level = level
+        self._read_fd, self._write_fd = os.pipe()
+        self._reader_pipe = os.fdopen(self._read_fd)
+        self.lines = []
+        self.start()
+
+    def fileno(self):
+        return self._write_fd
+
+    def run(self):
+        for line in iter(self._reader_pipe.readline, ""):
+            logger.log(self.level, line.strip("\n"))
+            self.lines.append(line)
+
+        self._reader_pipe.close()
+
+    def close(self):
+        os.close(self._write_fd)
+
+
+class CommandExecutor:
+    _logger = logger
+
+    def __init__(self, command: str):
+        self._command = command
+
+    def run(self, basepath: Path):
+        self._logger.debug(self._command)
+        out_logger = CommandLogger(logging.TRACE)
+        err_logger = CommandLogger(logging.TRACE)
+        try:
+            result = subprocess.run(
+                self._command.split(" "),
+                cwd=basepath,
+                stdout=out_logger,  # type: ignore
+                stderr=err_logger,  # type: ignore
+            )
+        finally:
+            out_logger.close()
+            err_logger.close()
+        if result.returncode != 0:
+            raise CommandError("".join(err_logger.lines or out_logger.lines))

project_manager/common/logging_colored.py

@@ -59,7 +59,7 @@ def addLoggingLevel(levelName, levelNum, methodName=None):
 addLoggingLevel("TRACE", logging.DEBUG - 5)
 
 
-class CustomFormatter(logging.Formatter):
+class ColoredFormatter(logging.Formatter):
     def __init__(self, fmt):
         super().__init__()
         self.fmt = fmt