feat: add manager, entry and command

project_manager/command.py

@@ -0,0 +1,39 @@
+__all__ = ("Command",)
+
+
+import logging
+import subprocess
+from pathlib import Path
+
+logger = logging.getLogger("command")
+
+
+class Command:
+    _logger = logger
+
+    def __init__(self, executor: str, *args: str) -> None:
+        self._executor = executor
+        self._args = args
+
+    def run(self, basepath: Path):
+        args = [self._executor, *self._args]
+        self._logger.info(" ".join(args))
+        result = subprocess.run(
+            args,
+            cwd=basepath,
+            stdout=subprocess.PIPE,
+            stderr=subprocess.PIPE,
+        )
+        self._logstd(result.stdout, logging.DEBUG)
+        if result.returncode != 0:
+            self._logstd(result.stderr, logging.ERROR)
+            raise RuntimeError(result.stderr)
+        else:
+            self._logstd(result.stderr, logging.WARNING)
+
+    @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)

project_manager/entry.py

@@ -0,0 +1,26 @@
+__all__ = ("Entry",)
+
+
+import logging
+from pathlib import Path
+from typing import Optional
+
+logger = logging.getLogger("entry")
+
+
+class Entry:
+    _logger = logger
+
+    def __init__(self, entry_type: str, basepath: Path):
+        self._type = entry_type
+        self._basepath = basepath
+
+    def setup(self):
+        raise NotImplemented
+
+    def __repr__(self):
+        return f"{self._type} {str(self._basepath)}"
+
+    @classmethod
+    def detect(cls, basepath: Path) -> Optional["Entry"]:
+        raise NotImplemented

project_manager/main.py

@@ -1,5 +1,37 @@
+import logging
+import os
+from argparse import ArgumentParser
+from pathlib import Path
+
+from project_manager.manager import ProjectManager
+
+
 def run():
-    print("project-manager")
+    parser = ArgumentParser()
+    parser.add_argument("task", nargs="?", default="show", choices=("show", "setup"))
+    parser.add_argument("-p", "--path", default=os.getcwd())
+    parser.add_argument("-i", "--ignore-errors", action="store_true")
+    parser.add_argument("-v", "--verbose", action="count", default=1)
+    parser.add_argument("-q", "--quiet", action="store_true")
+    args = parser.parse_args()
+
+    logging_level = logging.WARNING
+    if args.verbose > 1:
+        logging_level = logging.DEBUG
+    elif args.verbose > 0:
+        logging_level = logging.INFO
+    if args.quiet:
+        logging_level = logging.CRITICAL
+
+    logging.basicConfig(level=logging_level)
+
+    basepath = Path(args.path)
+    manager = ProjectManager(basepath, args.ignore_errors)
+
+    if args.task == "show":
+        manager.show()
+    elif args.task == "setup":
+        manager.setup()
 
 
 if __name__ == "__main__":

project_manager/manager.py

@@ -0,0 +1,32 @@
+import logging
+from pathlib import Path
+from typing import Iterable
+
+from .entry import Entry
+from .nodejs.entry import NodejsEntry
+from .python.entry import PythonEntry
+from .resolver import EntryDetector, EntryResolver
+
+__all__ = ("ProjectManager",)
+
+
+class ProjectManager:
+    def __init__(self, basepath: Path, ignore_entry_error: bool = False):
+        self._basepaht = basepath
+        self._resolver = EntryResolver(EntryDetector(PythonEntry, NodejsEntry))
+        self._ignore_entry_error = ignore_entry_error
+
+    def _iter_entries(self) -> Iterable[Entry]:
+        return self._resolver.resolve(self._basepaht)
+
+    def setup(self):
+        for entry in self._iter_entries():
+            try:
+                entry.setup()
+            except Exception as exc:
+                if not self._ignore_entry_error:
+                    raise exc
+
+    def show(self):
+        for entry in self._iter_entries():
+            print(entry)

project_manager/nodejs/command.py

@@ -0,0 +1,18 @@
+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/nodejs/entry.py

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

project_manager/python/command.py

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

@@ -0,0 +1,22 @@
+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)
+
+    @classmethod
+    def detect(cls, basepath: Path) -> Optional["PythonEntry"]:
+        if (basepath / "pyproject.toml").exists():
+            return cls(basepath)

project_manager/resolver.py

@@ -0,0 +1,34 @@
+from pathlib import Path
+from typing import Iterable, 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):
+        self._detector = detector
+
+    def resolve(self, basepath: Path, depth: int = 2) -> Iterable[Entry]:
+        if basepath.is_dir():
+            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)