Merge pull request #150 from lovit/feature/149

[#149] Pipeline - task template

.pre-commit-config.yaml

@@ -6,7 +6,8 @@ repos:
     hooks:
       - id: pyright
         entry: pyright
-        additional_dependencies: []
+        additional_dependencies:
+          - dacite==1.9.1
   - repo: https://github.com/astral-sh/ruff-pre-commit
     rev: v0.9.4
     hooks:

examples/dummy/pipeline.yaml

@@ -0,0 +1,4 @@
+pipeline:
+  - name: Dummy
+    args:
+      name: soynlp

examples/dummy/run.sh

@@ -0,0 +1 @@
+soynlp pipeline -c pipeline.yaml

pyproject.toml

@@ -9,6 +9,7 @@ soynlp = "soynlp.__main__:main"
 
 [tool.poetry.dependencies]
 python = "~3.12.8"
+dacite = "^1.9.1"
 
 [tool.poetry.group.dev.dependencies]
 pytest = "^8.3.0"

soynlp/cli.py

@@ -3,7 +3,7 @@
 from importlib.metadata import metadata
 from typing import Callable
 
-from soynlp.dummy import dummy
+from soynlp.pipeline.pipeline import Pipeline
 
 
 def main():
@@ -18,8 +18,9 @@ def main():
     parser.set_defaults(func=lambda: parser.print_usage())
     subparsers = parser.add_subparsers()
 
-    sp_dummy = subparsers.add_parser("dummy", help="dummy function")
-    sp_dummy.set_defaults(func=dummy)
+    sp_pipeline = subparsers.add_parser("pipeline", help="Run pipeline with config")
+    sp_pipeline.add_argument("-c", "--config_file", type=str, required=True, help="Config file path")
+    sp_pipeline.set_defaults(func=Pipeline.run)
 
     args = parser.parse_args()
     func = args.func

soynlp/configs/config.py

@@ -0,0 +1,25 @@
+from dataclasses import dataclass
+
+import dacite
+import yaml
+
+
+@dataclass
+class TaskConfig:
+    name: str
+    args: dict
+
+
+@dataclass
+class Config:
+    pipeline: list[TaskConfig]
+
+
+def from_yaml(path: str) -> Config:
+    with open(path) as file:
+        data = yaml.full_load(file)
+    return from_dict(data)  # type: ignore
+
+
+def from_dict(data: dict) -> Config:
+    return dacite.from_dict(data_class=Config, data=data)  # type: ignore

soynlp/pipeline/pipeline.py

@@ -0,0 +1,32 @@
+import importlib
+
+import dacite
+
+from soynlp.configs.config import Config, from_yaml
+from soynlp.pipeline.tasks import Task
+
+
+class Pipeline:
+    def __call__(self, config: Config):
+        tasks: list[Task] = self._load_tasks(config)
+        parameters: dict = {}
+
+        for task in tasks:
+            parameters |= task(parameters)
+        return parameters
+
+    def _load_tasks(self, config: Config) -> list[Task]:
+        tasks = []
+        for task_config in config.pipeline:
+            task_module = importlib.import_module("soynlp.pipeline.tasks")
+            task_class: type[Task] = getattr(task_module, f"{task_config.name}Task")
+            task_args = dacite.from_dict(data_class=task_class.args(), data=task_config.args)  # type: ignore
+            task = task_class(task_args)
+            tasks.append(task)
+        return tasks
+
+    @classmethod
+    def run(cls, config_file: str):
+        config = from_yaml(config_file)
+        pipeline = Pipeline()
+        print(pipeline(config))

soynlp/pipeline/tasks/__init__.py

@@ -0,0 +1,5 @@
+from soynlp.pipeline.tasks.dummy import DummyTask  # noqa F401
+from soynlp.pipeline.tasks.task import Task  # noqa F401
+
+
+__all__ = ("Task", "DummyTask")

soynlp/pipeline/tasks/dummy.py

@@ -0,0 +1,14 @@
+from dataclasses import dataclass
+
+from soynlp.pipeline.tasks.task import Task, TaskArgs
+
+
+@dataclass
+class DummyTaskArgs(TaskArgs):
+    name: str = "Dummy Task"
+
+
+class DummyTask(Task[DummyTaskArgs]):
+    def __call__(self, parameters: dict) -> dict:
+        print(f"Called in DummyTask({self._args.name})")
+        return parameters

soynlp/pipeline/tasks/task.py

@@ -0,0 +1,25 @@
+from abc import ABC, abstractmethod
+from dataclasses import dataclass
+from typing import Generic, TypeVar, get_args
+
+
+@dataclass
+class TaskArgs:
+    pass
+
+
+TaskArgsType = TypeVar("TaskArgsType", bound=TaskArgs)
+
+
+class Task(ABC, Generic[TaskArgsType]):
+    def __init__(self, args: TaskArgsType):
+        self._args = args
+
+    @classmethod
+    def args(cls) -> type[TaskArgs]:
+        generic_type = cls.__orig_bases__[0]  # type: ignore[attr-defined]
+        return get_args(generic_type)[0]
+
+    @abstractmethod
+    def __call__(self, parameters: dict) -> dict:
+        raise NotImplementedError