feat: rework default entries

project_manager/command.py

@@ -1,4 +1,7 @@
-__all__ = ("Command",)
+__all__ = (
+    "CommandError",
+    "Command",
+)
 
 
 import logging
@@ -8,6 +11,12 @@ from pathlib import Path
 logger = logging.getLogger(">")
 
 
+class CommandError(RuntimeError):
+    @property
+    def output(self):
+        return self.args[0].decode("utf-8").strip()
+
+
 class Command:
     _logger = logger
 
@@ -25,15 +34,12 @@ class Command:
             stderr=subprocess.PIPE,
         )
         self._logstd(result.stdout, logging.TRACE)
+        self._logstd(result.stderr, logging.TRACE)
         if result.returncode != 0:
-            self._logstd(result.stderr, logging.ERROR)
-            raise RuntimeError(result.stderr)
-        else:
-            self._logstd(result.stderr, logging.WARNING)
+            raise CommandError(result.stderr)
 
     @classmethod
     def _logstd(cls, data: bytes, level: int):
         if data:
-            for line in data.decode("utf-8").strip().splitlines():
-                if line:
-                    cls._logger.log(level, line)
+            output = data.decode("utf-8").strip()
+            cls._logger.log(level, f"\n{output}\n")

project_manager/entries/nodejs.py

@@ -1,20 +1,22 @@
 from pathlib import Path
 from typing import Optional
 
-from project_manager.nodejs.command import NpmCommand
-
 from ..entry import Entry
 
 
 class NodejsEntry(Entry):
     TYPE: str = "nodejs"
+    COMMANDS = {
+        "setup": [
+            ["npm", "install-clean"],
+        ],
+        "test": [
+            ["npm", "test"],
+        ],
+    }
 
     def __init__(self, basepath: Path):
-        super().__init__(self.TYPE, basepath)
-
-    def setup(self):
-        self._logger.info(f"setup {self}")
-        NpmCommand.install_clean().run(self._basepath)
+        super().__init__(self.TYPE, self.COMMANDS, basepath)
 
     @classmethod
     def detect(cls, basepath: Path) -> Optional["NodejsEntry"]:

project_manager/entries/python.py

@@ -0,0 +1,33 @@
+from pathlib import Path
+from typing import Optional
+
+from ..entry import Entry
+
+
+class PythonEntry(Entry):
+    TYPE: str = "python"
+    COMMANDS = {
+        "setup": [
+            ["poetry", "env", "use", "3.8"],
+            ["poetry", "install"],
+        ],
+        "lint": [
+            ["poetry", "run", "black", "--check", "--diff", "./"],
+            ["poetry", "run", "isort", "--profile", "black", "--check-only", "./"],
+        ],
+        "format": [
+            ["poetry", "run", "isort", "./"],
+            ["poetry", "run", "black", "./"],
+        ],
+        "test": [
+            ["poetry", "run", "pytest", "-p", "no:warnings"],
+        ],
+    }
+
+    def __init__(self, basepath: Path):
+        super().__init__(self.TYPE, self.COMMANDS, basepath)
+
+    @classmethod
+    def detect(cls, basepath: Path) -> Optional["PythonEntry"]:
+        if (basepath / "pyproject.toml").exists():
+            return cls(basepath)

project_manager/entry.py

@@ -6,7 +6,9 @@ __all__ = (
 
 import logging
 from pathlib import Path
-from typing import Optional
+from typing import Dict, List, Optional
+
+from .command import Command
 
 logger = logging.getLogger("entry")
 
@@ -18,24 +20,25 @@ class EntryCommandNotImplementedError(NotImplementedError):
 class Entry:
     _logger = logger
 
-    def __init__(self, entry_type: str, basepath: Path):
+    def __init__(
+        self,
+        entry_type: str,
+        commands: Dict[str, List[List[str]]],
+        basepath: Path,
+    ):
         self._type = entry_type
+        self._commands = commands
         self._basepath = basepath
 
-    def setup(self):
-        raise EntryCommandNotImplementedError("setup")
-
-    def lint(self):
-        raise EntryCommandNotImplementedError("lint")
-
-    def format(self):
-        raise EntryCommandNotImplementedError("format")
-
-    def test(self):
-        raise EntryCommandNotImplementedError("test")
-
-    def show(self):
-        self._logger.info(f"{self}")
+    def run_command(self, command_name: str):
+        if command_name == "show":
+            self._logger.info(f"{self}")
+            return
+        self._logger.info(f"{command_name} {self}")
+        if command_name not in self._commands:
+            raise EntryCommandNotImplementedError(command_name)
+        for command in self._commands[command_name]:
+            Command(*command).run(self._basepath)
 
     def __repr__(self):
         return f"{self._type} {str(self._basepath)}"

project_manager/logging.py

@@ -67,8 +67,8 @@ class CustomFormatter(logging.Formatter):
         super().__init__()
         self.fmt = fmt
         self.FORMATS = {
-            logging.TRACE: colored(self.fmt, "dark_grey"),
-            logging.DEBUG: colored(self.fmt, "grey"),
+            logging.TRACE: colored(self.fmt),
+            logging.DEBUG: colored(self.fmt, "white"),
             logging.INFO: colored(self.fmt, "blue"),
             logging.WARNING: colored(self.fmt, "yellow"),
             logging.ERROR: colored(self.fmt, "red"),

project_manager/main.py

@@ -1,5 +1,6 @@
 import logging
 import os
+import sys
 from argparse import ArgumentParser
 from ast import arg
 from pathlib import Path
@@ -31,9 +32,6 @@ def run():
     if args.quiet:
         logging_level = logging.CRITICAL
 
-    if logging_level <= logging.INFO:
-        logging.getLogger(">").setLevel(logging.CRITICAL)
-
     stdout_handler = logging.StreamHandler()
     stdout_handler.setFormatter(CustomFormatter("%(message)s"))
 
@@ -45,7 +43,9 @@ def run():
 
     basepath = Path(args.path)
     manager = ProjectManager(basepath)
-    manager.run_command(args.task)
+    result = manager.run_command(args.task)
+    if result["failed"] > 0:
+        sys.exit(1)
 
 
 if __name__ == "__main__":

project_manager/manager.py

@@ -3,9 +3,10 @@ from collections import Counter
 from pathlib import Path
 from typing import Dict, Iterable
 
+from .command import CommandError
+from .entries.nodejs import NodejsEntry
+from .entries.python import PythonEntry
 from .entry import Entry, EntryCommandNotImplementedError
-from .nodejs.entry import NodejsEntry
-from .python.entry import PythonEntry
 from .resolver import EntryDetector, EntryResolver
 
 __all__ = ("ProjectManager",)
@@ -18,20 +19,25 @@ class ProjectManager:
 
     def __init__(self, basepath: Path):
         self._basepaht = basepath
-        self._resolver = EntryResolver(EntryDetector(PythonEntry, NodejsEntry))
+        self._resolver = EntryResolver(
+            EntryDetector(PythonEntry, NodejsEntry), ["node_modules", ".venv", ".git"]
+        )
 
     def _iter_entries(self) -> Iterable[Entry]:
         return self._resolver.resolve(self._basepaht)
 
-    def run_command(self, command: str) -> Dict[str, int]:
+    def run_command(self, command_name: str) -> Dict[str, int]:
         counter = Counter()
         for entry in self._iter_entries():
             try:
-                getattr(entry, command)()
+                entry.run_command(command_name)
                 counter["success"] += 1
             except EntryCommandNotImplementedError:
-                self._logger.warn(f"{command} {entry}")
+                self._logger.warn(f"{command_name} {entry}")
                 counter["skipped"] += 1
+            except CommandError as error:
+                self._logger.error(f"{entry}\n{error.output}")
+                counter["failed"] += 1
             except Exception:
                 self._logger.exception(entry)
                 counter["failed"] += 1

project_manager/nodejs/command.py

@@ -1,18 +0,0 @@
-from ..command import Command
-
-__all__ = ("NpmCommand",)
-
-
-class NpmCommand(Command):
-    EXECUTOR: str = "npm"
-
-    def __init__(self, *args: str) -> None:
-        super().__init__(self.EXECUTOR, *args)
-
-    @classmethod
-    def install(cls) -> "NpmCommand":
-        return cls("install")
-
-    @classmethod
-    def install_clean(cls) -> "NpmCommand":
-        return cls("install-clean")

project_manager/python/command.py

@@ -1,18 +0,0 @@
-from ..command import Command
-
-__all__ = ("PoetryCommand",)
-
-
-class PoetryCommand(Command):
-    EXECUTOR: str = "poetry"
-
-    def __init__(self, *args: str) -> None:
-        super().__init__(self.EXECUTOR, *args)
-
-    @classmethod
-    def env_use(cls, python_version: str) -> "PoetryCommand":
-        return cls("env", "use", python_version)
-
-    @classmethod
-    def install(cls) -> "PoetryCommand":
-        return cls("install")

project_manager/python/entry.py

@@ -1,34 +0,0 @@
-from pathlib import Path
-from typing import Optional
-
-from ..entry import Entry
-from .command import PoetryCommand
-
-
-class PythonEntry(Entry):
-    TYPE: str = "python"
-
-    def __init__(self, basepath: Path):
-        super().__init__(self.TYPE, basepath)
-
-    def setup(self):
-        self._logger.info(f"setup {self}")
-        PoetryCommand.env_use("3.8").run(self._basepath)
-        PoetryCommand.install().run(self._basepath)
-
-    def lint(self):
-        self._logger.info(f"lint {self}")
-        PoetryCommand("run", "black", "--check", "--diff", "./").run(self._basepath)
-        PoetryCommand("run", "isort", "--profile", "black", "--check-only", "./").run(
-            self._basepath
-        )
-
-    def format(self):
-        self._logger.info(f"format {self}")
-        PoetryCommand("run", "isort", "./").run(self._basepath)
-        PoetryCommand("run", "black", "./").run(self._basepath)
-
-    @classmethod
-    def detect(cls, basepath: Path) -> Optional["PythonEntry"]:
-        if (basepath / "pyproject.toml").exists():
-            return cls(basepath)

project_manager/resolver.py

@@ -1,5 +1,5 @@
 from pathlib import Path
-from typing import Iterable, Optional, Type
+from typing import Iterable, List, Optional, Type
 
 from .entry import Entry
 
@@ -21,11 +21,14 @@ class EntryDetector:
 
 
 class EntryResolver:
-    def __init__(self, detector: EntryDetector):
+    def __init__(self, detector: EntryDetector, ignore: List[str]):
         self._detector = detector
+        self._ignore = ignore
 
-    def resolve(self, basepath: Path, depth: int = 2) -> Iterable[Entry]:
+    def resolve(self, basepath: Path, depth: int = 128) -> Iterable[Entry]:
         if basepath.is_dir():
+            if basepath.name in self._ignore:
+                return
             entry = self._detector.detect(basepath)
             if entry:
                 yield entry