feat: add presets

project_manager/command.py

@@ -1,6 +1,6 @@
 __all__ = (
     "CommandError",
-    "Command",
+    "CommandExecutor",
 )
 
 
@@ -8,7 +8,7 @@ import logging
 import subprocess
 from pathlib import Path
 
-logger = logging.getLogger(">")
+logger = logging.getLogger("command")
 
 
 class CommandError(RuntimeError):
@@ -17,18 +17,16 @@ class CommandError(RuntimeError):
         return self.args[0].decode("utf-8").strip()
 
 
-class Command:
+class CommandExecutor:
     _logger = logger
 
-    def __init__(self, executor: str, *args: str) -> None:
-        self._executor = executor
-        self._args = args
+    def __init__(self, command: str) -> None:
+        self._command = command
 
     def run(self, basepath: Path):
-        args = [self._executor, *self._args]
-        self._logger.debug(" ".join(args))
+        self._logger.debug(self._command)
         result = subprocess.run(
-            args,
+            self._command.split(" "),
             cwd=basepath,
             stdout=subprocess.PIPE,
             stderr=subprocess.PIPE,

project_manager/default_preset.toml

@@ -0,0 +1,35 @@
+[presets.python]
+filename = "pyproject.toml"
+loader = "tomli:load"
+
+[presets.python.tasks.setup.poetry]
+conditions = { "build-system.build-backend" = "poetry.core.masonry.api" }
+commands = ["poetry env use 3.8", "poetry install"]
+
+[presets.python.tasks.test.pytest]
+conditions = { "tool.poetry.group.dev.dependencies.pytest" = "*" }
+commands = ["poetry run pytest -p no:warnings"]
+
+[presets.python.tasks.lint.black]
+conditions = { "tool.poetry.group.dev.dependencies.black" = "*" }
+commands = ["poetry run black --check --diff ./"]
+
+[presets.python.tasks.format.black]
+conditions = { "tool.poetry.group.dev.dependencies.black" = "*" }
+commands = ["poetry run black ./"]
+
+[presets.python.tasks.lint.isort]
+conditions = { "tool.poetry.group.dev.dependencies.isort" = "*" }
+commands = ["poetry run isort --profile black --check-only ./"]
+
+[presets.python.tasks.format.isort]
+conditions = { "tool.poetry.group.dev.dependencies.isort" = "*" }
+commands = ["poetry run isort ./"]
+
+[presets.nodejs]
+filename = "package.json"
+loader = "json:load"
+
+[presets.nodejs.tasks.setup.npm]
+conditions = {}
+commands = ["npm install-clean"]

project_manager/entries/nodejs.py

@@ -1,24 +0,0 @@
-from pathlib import Path
-from typing import Optional
-
-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, self.COMMANDS, basepath)
-
-    @classmethod
-    def detect(cls, basepath: Path) -> Optional["NodejsEntry"]:
-        if (basepath / "package.json").exists():
-            return cls(basepath)

project_manager/entries/python.py

@@ -1,33 +0,0 @@
-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

@@ -1,19 +1,20 @@
-__all__ = (
-    "EntryCommandNotImplementedError",
-    "Entry",
-)
-
-
 import logging
 from pathlib import Path
-from typing import Dict, List, Optional
+from typing import Dict, Iterable, List, Optional
 
-from .command import Command
+from .command import CommandExecutor
+
+__all__ = (
+    "EntryTaskNotImplementedError",
+    "Entry",
+    "EntryDetector",
+    "EntryResolver",
+)
 
 logger = logging.getLogger("entry")
 
 
-class EntryCommandNotImplementedError(NotImplementedError):
+class EntryTaskNotImplementedError(NotImplementedError):
     pass
 
 
@@ -22,27 +23,50 @@ class Entry:
 
     def __init__(
         self,
-        entry_type: str,
-        commands: Dict[str, List[List[str]]],
+        tasks: Dict[str, List[str]],
         basepath: Path,
     ):
-        self._type = entry_type
-        self._commands = commands
+        self._tasks = tasks
         self._basepath = basepath
 
-    def run_command(self, command_name: str):
-        if command_name == "show":
-            self._logger.info(f"{self}")
+    def show(self):
+        self._logger.info(f"{self}")
+        for task_name, commands in self._tasks.items():
+            self._logger.debug(f"  {task_name}")
+            for command in commands:
+                self._logger.trace(f"     {command}")
+
+    def run_task(self, task_name: str):
+        if task_name == "show":
+            self.show()
             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)
+        if task_name not in self._tasks:
+            raise EntryTaskNotImplementedError(task_name)
+        self._logger.info(f"{task_name} {self}")
+        for command in self._tasks[task_name]:
+            CommandExecutor(command).run(self._basepath)
 
     def __repr__(self):
-        return f"{self._type} {str(self._basepath)}"
+        return f"{str(self._basepath)}"
 
-    @classmethod
-    def detect(cls, basepath: Path) -> Optional["Entry"]:
+
+class EntryDetector:
+    def detect_entry(self, basepath: Path) -> Optional["Entry"]:
         raise NotImplemented
+
+
+class EntryResolver:
+    def __init__(self, detector: EntryDetector, ignore: List[str]):
+        self._detector = detector
+        self._ignore = ignore
+
+    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_entry(basepath)
+            if entry:
+                yield entry
+            if depth > 0 and basepath.is_dir():
+                for path in basepath.iterdir():
+                    yield from self.resolve(path, depth - 1)

project_manager/logging.py

@@ -76,6 +76,6 @@ class CustomFormatter(logging.Formatter):
         }
 
     def format(self, record: logging.LogRecord):
-        log_fmt = colored(record.name, "green") + " " + self.FORMATS.get(record.levelno)
+        log_fmt = self.FORMATS.get(record.levelno)
         formatter = logging.Formatter(log_fmt)
         return formatter.format(record)

project_manager/main.py

@@ -2,7 +2,6 @@ import logging
 import os
 import sys
 from argparse import ArgumentParser
-from ast import arg
 from pathlib import Path
 
 from project_manager.logging import CustomFormatter
@@ -43,7 +42,7 @@ def run():
 
     basepath = Path(args.path)
     manager = ProjectManager(basepath)
-    result = manager.run_command(args.task)
+    result = manager.run_task(args.task)
     if result["failed"] > 0:
         sys.exit(1)
 

project_manager/manager.py

@@ -4,10 +4,8 @@ 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 .resolver import EntryDetector, EntryResolver
+from .entry import Entry, EntryResolver, EntryTaskNotImplementedError
+from .preset.entry import PresetEntryDetector
 
 __all__ = ("ProjectManager",)
 
@@ -20,20 +18,21 @@ class ProjectManager:
     def __init__(self, basepath: Path):
         self._basepaht = basepath
         self._resolver = EntryResolver(
-            EntryDetector(PythonEntry, NodejsEntry), ["node_modules", ".venv", ".git"]
+            PresetEntryDetector(Path(__file__).parent / "default_preset.toml"),
+            ["node_modules", ".venv", ".git"],
         )
 
     def _iter_entries(self) -> Iterable[Entry]:
         return self._resolver.resolve(self._basepaht)
 
-    def run_command(self, command_name: str) -> Dict[str, int]:
+    def run_task(self, task_name: str) -> Dict[str, int]:
         counter = Counter()
         for entry in self._iter_entries():
             try:
-                entry.run_command(command_name)
+                entry.run_task(task_name)
                 counter["success"] += 1
-            except EntryCommandNotImplementedError:
-                self._logger.warn(f"{command_name} {entry}")
+            except EntryTaskNotImplementedError:
+                self._logger.warn(f"{task_name} {entry}")
                 counter["skipped"] += 1
             except CommandError as error:
                 self._logger.error(f"{entry}\n{error.output}")

project_manager/preset/entry.py

@@ -0,0 +1,28 @@
+from collections import defaultdict
+from pathlib import Path
+from typing import Optional
+
+from ..entry import Entry, EntryDetector
+from .util import PresetUtil
+
+__all__ = ("PresetEntryDetector",)
+
+
+class PresetEntryDetector(EntryDetector):
+    def __init__(self, filename: Path):
+        self._preset = PresetUtil.resolve_preset(filename)
+
+    def detect_entry(self, basepath: Path) -> Optional["Entry"]:
+        tasks = defaultdict(list)
+        for preset_name, preset in self._preset.presets.items():
+            filename = basepath / preset.filename
+            if filename.exists():
+                loader = PresetUtil.resolve_loader(preset.loader)
+                with open(filename, "rb") as fd:
+                    data = loader(fd)
+                for task_name, task_pack in preset.tasks.items():
+                    for task_item_name, task in task_pack.items():
+                        if PresetUtil.check_conditions(data, task.conditions):
+                            tasks[task_name] += task.commands
+        if tasks:
+            return Entry(tasks, basepath)

project_manager/preset/model.py

@@ -0,0 +1,20 @@
+from typing import Dict, List
+
+from pydantic import BaseModel
+
+__all__ = ("TaskPreset", "FilePreset", "Preset")
+
+
+class TaskPreset(BaseModel):
+    conditions: Dict[str, str]
+    commands: List[str]
+
+
+class FilePreset(BaseModel):
+    filename: str
+    loader: str
+    tasks: Dict[str, Dict[str, TaskPreset]]
+
+
+class Preset(BaseModel):
+    presets: Dict[str, FilePreset]

project_manager/preset/util.py

@@ -0,0 +1,45 @@
+__all__ = ("PresetUtil",)
+
+import importlib
+from pathlib import Path
+from typing import Any, Dict
+
+import tomli
+
+from .model import Preset
+
+
+class PresetUtil:
+    @classmethod
+    def resolve_loader(cls, loader: str) -> Any:
+        module_name, item_name = loader.split(":")
+        module = importlib.import_module(module_name)
+        return getattr(module, item_name)
+
+    @classmethod
+    def get_key_path_value(cls, key_path: str, data: dict):
+        path = key_path.split(".")
+        value = data.get(path[0])
+        for key in path[1:]:
+            if value:
+                value = value.get(key)
+        return value
+
+    @classmethod
+    def match_condition_value(cls, value: Any, pattern: str) -> bool:
+        if pattern == "*":
+            return bool(value)
+        return value == pattern
+
+    @classmethod
+    def check_conditions(cls, data: dict, conditions: Dict[str, str]) -> bool:
+        for key, value in conditions.items():
+            if not cls.match_condition_value(cls.get_key_path_value(key, data), value):
+                return False
+        return True
+
+    @classmethod
+    def resolve_preset(cls, filename: Path) -> Preset:
+        with open(filename, "rb") as fd:
+            preset_data = tomli.load(fd)
+        return Preset(**preset_data)

project_manager/resolver.py

@@ -1,37 +0,0 @@
-from pathlib import Path
-from typing import Iterable, List, Optional, Type
-
-from .entry import Entry
-
-__all__ = (
-    "EntryDetector",
-    "EntryResolver",
-)
-
-
-class EntryDetector:
-    def __init__(self, *entry_types: Type[Entry]):
-        self._entry_types = entry_types
-
-    def detect(self, basepath: Path) -> Optional[Entry]:
-        for entry_type in self._entry_types:
-            entry = entry_type.detect(basepath)
-            if entry:
-                return entry
-
-
-class EntryResolver:
-    def __init__(self, detector: EntryDetector, ignore: List[str]):
-        self._detector = detector
-        self._ignore = ignore
-
-    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
-            if depth > 0 and basepath.is_dir():
-                for path in basepath.iterdir():
-                    yield from self.resolve(path, depth - 1)