refactor: update project code

project_manager/__init__.py

@@ -1 +0,0 @@
-from .common.logging_custom import *

project_manager/command.py

@@ -47,7 +47,7 @@ class CommandLogger(threading.Thread):
 class CommandExecutor:
     _logger = logger
 
-    def __init__(self, command: str) -> None:
+    def __init__(self, command: str):
         self._command = command
 
     def run(self, basepath: Path):
@@ -58,8 +58,8 @@ class CommandExecutor:
             result = subprocess.run(
                 self._command.split(" "),
                 cwd=basepath,
-                stdout=out_logger,
-                stderr=err_logger,
+                stdout=out_logger,  # type: ignore
+                stderr=err_logger,  # type: ignore
             )
         finally:
             out_logger.close()

project_manager/entry.py

@@ -1,6 +1,6 @@
 import logging
 from pathlib import Path
-from typing import Dict, Iterable, List, Optional
+from typing import Iterable
 
 from .command import CommandExecutor
 
@@ -23,18 +23,18 @@ class Entry:
 
     def __init__(
         self,
-        tasks: Dict[str, List[str]],
+        tasks: dict[str, list[str]],
         basepath: Path,
     ):
         self._tasks = tasks
         self._basepath = basepath
 
     def show(self):
-        self._logger.info(f"{self}")
+        self._logger.info("%s", self)
         for task_name, commands in self._tasks.items():
-            self._logger.info(f"  {task_name}")
+            self._logger.info("  %s", task_name)
             for command in commands:
-                self._logger.debug(f"     {command}")
+                self._logger.debug("     %s", command)
 
     def run_task(self, task_name: str):
         if task_name == "show":
@@ -42,7 +42,7 @@ class Entry:
             return
         if task_name not in self._tasks:
             raise EntryTaskNotImplementedError(task_name)
-        self._logger.info(f"{task_name} {self}")
+        self._logger.info("%s %s", task_name, self)
         for command in self._tasks[task_name]:
             CommandExecutor(command).run(self._basepath)
 
@@ -51,12 +51,12 @@ class Entry:
 
 
 class EntryDetector:
-    def detect_entry(self, basepath: Path) -> Optional["Entry"]:
-        raise NotImplemented
+    def detect_entry(self, basepath: Path) -> Entry | None:
+        raise NotImplementedError
 
 
 class EntryResolver:
-    def __init__(self, detector: EntryDetector, ignore: List[str]):
+    def __init__(self, detector: EntryDetector, ignore: list[str]):
         self._detector = detector
         self._ignore = ignore
 

project_manager/manager.py

@@ -1,7 +1,7 @@
 import logging
 from collections import Counter
 from pathlib import Path
-from typing import Dict, Iterable
+from typing import Iterable
 
 from .command import CommandError
 from .entry import Entry, EntryResolver, EntryTaskNotImplementedError
@@ -26,20 +26,20 @@ class ProjectManager:
     def _iter_entries(self) -> Iterable[Entry]:
         return self._resolver.resolve(self._basepaht)
 
-    def run_task(self, task_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_task(task_name)
                 counter["success"] += 1
             except EntryTaskNotImplementedError:
-                self._logger.warn(f"{task_name} {entry}")
+                self._logger.warning("%s %s", task_name, entry)
                 counter["skipped"] += 1
             except CommandError as error:
-                self._logger.error(f"{entry}\n{error.output}")
+                self._logger.error("%s\n%s", entry, error.output)
                 counter["failed"] += 1
-            except Exception:
+            except:
                 self._logger.exception(entry)
                 counter["failed"] += 1
-        self._logger.info(f"result {dict(counter)}")
+        self._logger.info("result %s", dict(counter))
         return counter

project_manager/pipe/core.py

@@ -1,7 +1,7 @@
 import importlib
 import re
 from pathlib import Path
-from typing import Any, Generic, List, Optional, Type, TypeVar
+from typing import Any, Generic, TypeVar
 
 __all__ = ("Pipe", "PipeRegistry", "PipeParser", "PipeRunner", "PIPE_REGISTRY")
 
@@ -12,7 +12,7 @@ O = TypeVar("O")
 class Pipe(Generic[I, O]):
     type: str
 
-    def apply(self, value: I, basepath: Path) -> Optional[O]:
+    def apply(self, value: I, basepath: Path) -> O | None:
         raise NotImplementedError
 
 
@@ -20,12 +20,13 @@ class PipeRegistry:
     def __init__(self):
         self._pipes = {}
 
-    def register(self, pipe_type: Type[Pipe]):
+    def register(self, pipe_type: type[Pipe]):
         self._pipes[pipe_type.type] = pipe_type
 
     def register_module(self, module_name: str):
         module = importlib.import_module(module_name)
         modulepath = Path(module.__file__).parent
+        # pylint:disable=too-many-nested-blocks
         for item in modulepath.iterdir():
             if item.suffix == ".py" and not item.name.startswith("__"):
                 sub_module = importlib.import_module(f"{module_name}.{item.stem}")
@@ -35,14 +36,14 @@ class PipeRegistry:
                             if baseclass == Pipe:
                                 self.register(v)
 
-    def get(self, pipe_type: str) -> Type[Pipe]:
+    def get(self, pipe_type: str) -> type[Pipe]:
         return self._pipes[pipe_type]
 
 
 class PipeParser:
     _PIPE_PATTERN = re.compile("^(\\w+)\\((.*)\\)")
 
-    def parse(self, pipeline: str) -> List[Pipe]:
+    def parse(self, pipeline: str) -> list[Pipe]:
         chain = [item.strip() for item in pipeline.split("|")]
         result = []
         for part in chain:
@@ -59,11 +60,11 @@ class PipeParser:
 
 
 class PipeRunner:
-    def __init__(self, basepath: Path, parser: Optional[PipeParser] = None):
+    def __init__(self, basepath: Path, parser: PipeParser | None = None):
         self._basepath = basepath
         self._parser = parser or PipeParser()
 
-    def run(self, pipes: List[Pipe]) -> Any:
+    def run(self, pipes: list[Pipe]) -> Any:
         result = None
         for pipe in pipes:
             result = pipe.apply(result, self._basepath)
@@ -75,7 +76,7 @@ class PipeRunner:
         pipe = self._parser.parse(pipeline)
         return self.run(pipe)
 
-    def check_conditions(self, conditions: Optional[List[str]]) -> bool:
+    def check_conditions(self, conditions: list[str] | None) -> bool:
         if conditions:
             for condition in conditions:
                 if not self.resolve_value(condition):

project_manager/pipe/type/file.py

@@ -1,5 +1,5 @@
 from pathlib import Path
-from typing import Any, Optional
+from typing import Any
 
 from ...util import reflect
 from ..core import Pipe
@@ -16,7 +16,7 @@ class FileData:
         ".json": "json:load",
     }
 
-    def __init__(self, filepath: Path, loader: Optional[str] = None):
+    def __init__(self, filepath: Path, loader: str | None = None):
         self._filepath = filepath
         self._loader = loader
         self._data = None
@@ -38,6 +38,7 @@ class FileData:
             self._load_data()
         if self._data:
             return self._data.get(key)
+        return None
 
     def __bool__(self):
         return self._filepath.exists()
@@ -46,11 +47,11 @@ class FileData:
 class FilePipe(Pipe[None, Any]):
     type = "file"
 
-    def __init__(self, filename: Optional[str], loader: Optional[str] = None):
+    def __init__(self, filename: str | None, loader: str | None = None):
         self._filename = filename
         self._loader = loader
 
-    def apply(self, value: Optional[str], basepath: Path) -> Optional[Any]:
+    def apply(self, value: str | None, basepath: Path) -> Any | None:
         filepath = Path(value or self._filename)  # TODO: raise none filepath exception
         if not filepath.is_absolute():
             filepath = basepath / filepath

project_manager/pipe/type/get.py

@@ -1,5 +1,5 @@
 from pathlib import Path
-from typing import Any, Optional
+from typing import Any
 
 from ..core import Pipe
 
@@ -12,7 +12,7 @@ class GetPipe(Pipe[dict, Any]):
     def __init__(self, keypath: str):
         self._keypath = keypath
 
-    def apply(self, value: dict, basepath: Path) -> Optional[Any]:
+    def apply(self, value: dict, basepath: Path) -> Any | None:
         path = self._keypath.split(".")
         result = value.get(path[0])
         for key in path[1:]:
@@ -24,9 +24,9 @@ class GetPipe(Pipe[dict, Any]):
 class KeyPipe(Pipe[dict, Any]):
     type = "key"
 
-    def __init__(self, keyindex: Optional[str]):
+    def __init__(self, keyindex: str | None):
         self._keyindex = int(keyindex) if keyindex else 0
 
-    def apply(self, value: dict, basepath: Path) -> Optional[Any]:
+    def apply(self, value: dict, basepath: Path) -> Any | None:
         keys = list(value.keys())
         return keys[self._keyindex]

project_manager/pipe/type/search.py

@@ -1,6 +1,5 @@
 import re
 from pathlib import Path
-from typing import Optional
 
 from ..core import Pipe
 
@@ -14,7 +13,8 @@ class SearchPipe(Pipe[str, str]):
         self._patter = patter
         self._position = int(position)
 
-    def apply(self, value: str, basepath: Path) -> Optional[str]:
+    def apply(self, value: str, basepath: Path) -> str | None:
         matched = re.compile(self._patter).search(value)
         if matched:
             return matched.group(self._position)
+        return None

project_manager/preset/entry.py

@@ -1,6 +1,5 @@
 from collections import defaultdict
 from pathlib import Path
-from typing import Optional
 
 from ..entry import Entry, EntryDetector
 from ..pipe.core import PipeParser, PipeRunner
@@ -12,21 +11,21 @@ __all__ = ("PresetEntryDetector",)
 class PresetEntryDetector(EntryDetector):
     _PIPE_PARSER = PipeParser()
 
-    def __init__(self, preset_resolver_config: Optional[PresetResolverConfig] = None):
+    def __init__(self, preset_resolver_config: PresetResolverConfig | None = None):
         self._pipeparser = PipeParser()
         self._preset_resolver_config = preset_resolver_config
 
-    def detect_entry(self, basepath: Path) -> Optional["Entry"]:
+    def detect_entry(self, basepath: Path) -> Entry | None:
         preset = PresetResolver(basepath, self._preset_resolver_config).resolve_preset()
         if preset.skip:
-            return
+            return None
         piperunner = PipeRunner(basepath)
         tasks = defaultdict(list)
-        for preset_name, preset in preset.presets.items():
+        for preset in preset.presets.values():
             if not piperunner.check_conditions(preset.conditions):
                 continue
             for task_name, task_pack in preset.tasks.items():
-                for task_item_name, task in task_pack.items():
+                for task in task_pack.values():
                     if piperunner.check_conditions(task.conditions):
                         variables = {
                             "entry_path": basepath,
@@ -43,3 +42,4 @@ class PresetEntryDetector(EntryDetector):
                         ]
         if tasks:
             return Entry(tasks, basepath)
+        return None

project_manager/preset/resolve.py

@@ -1,6 +1,6 @@
 from dataclasses import dataclass
 from pathlib import Path
-from typing import Dict, Iterable, Optional
+from typing import Iterable
 
 import tomli
 
@@ -21,16 +21,15 @@ class PresetResolverConfig:
 
 
 class PresetResolver:
-    _source_cache: Dict[Path, dict] = {}
+    _source_cache: dict[Path, dict] = {}
 
-    def __init__(self, basepath: Path, config: Optional[PresetResolverConfig] = None):
+    def __init__(self, basepath: Path, config: PresetResolverConfig | None = None):
         self._basepath = basepath
         self._config = config or PresetResolverConfig()
 
     def iter_sources(self) -> Iterable[Path]:
         if self._config.default:
-            for filepath in (Path(__file__).parent / "default").iterdir():
-                yield filepath
+            yield from (Path(__file__).parent / "default").iterdir()
         if self._config.user:
             user_preset = Path.home() / ".preset.toml"
             if user_preset.exists():