refactor: add pipe module

project_manager/preset/entry.py

@@ -1,30 +1,34 @@
+from cmath import pi
 from collections import defaultdict
 from pathlib import Path
 from typing import Optional
 
 from ..entry import Entry, EntryDetector
+from ..pipe.core import PipeParser, PipeRunner
 from .util import PresetUtil
-from .value import ValueResolver
 
 __all__ = ("PresetEntryDetector",)
 
 
 class PresetEntryDetector(EntryDetector):
+    _PIPE_PARSER = PipeParser()
+
     def __init__(self, filename: Path):
         self._preset = PresetUtil.resolve_preset(filename)
+        self._pipeparser = PipeParser()
 
     def detect_entry(self, basepath: Path) -> Optional["Entry"]:
+        piperunner = PipeRunner(basepath)
         tasks = defaultdict(list)
         for preset_name, preset in self._preset.presets.items():
             filename = basepath / preset.filename
             if filename.exists():
-                value_resolver = ValueResolver(basepath)
                 for task_name, task_pack in preset.tasks.items():
                     for task_item_name, task in task_pack.items():
-                        if value_resolver.check_conditions(task.conditions):
+                        if piperunner.check_conditions(task.conditions):
                             variables = (
                                 {
-                                    k: value_resolver.resolve_value(v)
+                                    k: piperunner.resolve_value(v)
                                     for k, v in task.variables.items()
                                 }
                                 if task.variables

project_manager/preset/model.py

@@ -2,15 +2,12 @@ from typing import Dict, List, Optional
 
 from pydantic import BaseModel
 
-__all__ = ("PipeStr", "TaskPreset", "FilePreset", "Preset")
-
-
-PipeStr = str
+__all__ = ("TaskPreset", "FilePreset", "Preset")
 
 
 class TaskPreset(BaseModel):
-    conditions: List[PipeStr]
-    variables: Optional[Dict[str, PipeStr]]
+    conditions: List[str]
+    variables: Optional[Dict[str, str]]
     commands: List[str]
 
 

project_manager/preset/value.py

@@ -1,88 +0,0 @@
-import importlib
-import re
-from pathlib import Path
-from typing import Any, List, Optional
-
-from .model import PipeStr
-
-__all__ = ("ValueResolver",)
-
-
-class ValueResolver:
-    PIPE_PATTERN = re.compile("^(\\w+)\\((.*)\\)")
-    FILE_LOADERS = {
-        ".toml": "tomli:load",
-        ".json": "json:load",
-    }
-
-    def __init__(self, basepath: Path):
-        self._basepath = basepath
-
-    def _reflect(self, value: str) -> Any:
-        if ":" in value:
-            module_name, item_name = value.split(":")
-        else:
-            module_name, item_name = "builtins", value
-        module = importlib.import_module(module_name)
-        return getattr(module, item_name)
-
-    def pipe_file(self, pipe, filename: str, loader: Optional[str] = None):
-        filepath = Path(filename)
-        if not filepath.is_absolute():
-            filepath = self._basepath / filepath
-        if loader is None:
-            loader = self.FILE_LOADERS.get(Path(filename).suffix)
-        if not filepath.exists():
-            return None
-        with open(filepath, "rb") as fd:
-            if loader:
-                loader_fun = self._reflect(loader)
-                data = loader_fun(fd)
-            else:
-                data = fd.read()
-        return data
-
-    def pipe_get(self, pipe, keypath: str):
-        path = keypath.split(".")
-        value = pipe.get(path[0])
-        for key in path[1:]:
-            if value:
-                value = value.get(key)
-        return value
-
-    def pipe_search(self, pipe, pattern: str, position_arg: str = "1"):
-        matched = re.compile(pattern).search(pipe)
-        position = int(position_arg)
-        if matched:
-            return matched.group(position)
-
-    def pipe_equal(self, pipe, value: Any):
-        return pipe == value
-
-    def pipe_convert(self, pipe, converter: str):
-        converter_fun = self._reflect(converter)
-        return converter_fun(pipe)
-
-    def pipe_not(self, pipe):
-        return not pipe
-
-    def resolve_value(self, operator: str) -> Any:
-        chain = [item.strip() for item in operator.split("|")]
-        pipe = None
-        for part in chain:
-            matched = self.PIPE_PATTERN.match(part)
-            if matched:
-                pipe_name = matched.group(1)
-                pipe_args = [item.strip() for item in matched.group(2).split(",")]
-            else:
-                pipe_name = part.strip()
-                pipe_args = []
-            pipe_func = getattr(self, f"pipe_{pipe_name}")
-            pipe = pipe_func(pipe, *pipe_args)
-        return pipe
-
-    def check_conditions(self, conditions: List[PipeStr]) -> bool:
-        for condition in conditions:
-            if not self.resolve_value(condition):
-                return False
-        return True