PaeoniaCommon/SkillsOdea

建立流程

所有路徑皆相對於儲存庫根目錄。套件位於 process_package/。

父類別如何運作

process_package/toy_process/process.py 定義了 Process：

• __init__(**overrides) 設定 self.params = {**self.set_parameters(), **overrides}， 然後 self.steps = self.set_steps()。
• set_parameters() 和 set_steps() 是抽象方法。缺少其中任何一個的子類別 都無法被實例化。
• run(context=None) 依序迭代 self.steps：
  • 如果是 Process 實例 → step.run(context)，巢狀執行，使用同一個 context 字典
  • 如果是任何其他可呼叫物件 → step(context)
  • 其他任何東西 → TypeError
• run 會回傳 context 字典，並印出帶有縮排的追蹤資訊。

請勿修改 toy_process/，或在子類別中覆寫 __init__ 或 run。

子類別的規則

1. 從 from toy_process import Process 匯入並繼承它。
2. set_parameters(self) 回傳一個包含預設值的 dict。如果沒有參數，則使用 {}。
3. set_steps(self) 回傳一個 list。每個項目可以是：
   • 一個綁定方法：self.count_words，不加括號，或
   • 一個 Process 實例：CleanText(lowercase=self.params["lowercase"])。
4. 每個步驟方法都有簽章 def name(self, context):。
   • 從 self.params["..."] 讀取設定。
   • 在 context["..."] 中讀取和寫入資料。
   • 不回傳任何東西。回傳值會被忽略。
5. 在使用巢狀流程類別的類別上方定義它，或直接匯入它。

範本

"""流程的一行描述。

從 process_package/ 執行：
    python -m examples.<模組名稱>
"""

from toy_process import Process


class SubTask(Process):
    def set_parameters(self):
        return {"option": True}

    def set_steps(self):
        return [self.do_part]

    def do_part(self, context):
        if self.params["option"]:
            context["value"] = context["value"] * 2


class MainTask(Process):
    def set_parameters(self):
        return {"start": 1, "option": True}

    def set_steps(self):
        return [
            self.load,
            SubTask(option=self.params["option"]),  # 巢狀子流程
            self.finish,
        ]

    def load(self, context):
        context["value"] = self.params["start"]

    def finish(self, context):
        context["result"] = f"value is {context['value']}"


if __name__ == "__main__":
    print(MainTask(start=5).run()["result"])

需要建立的檔案

1. process_package/examples/<模組名稱>.py：根據範本建立的流程。
2. process_package/tests/test_<模組名稱>.py：一個 unittest 測試，呼叫 .run() 並驗證回傳 context 中的鍵。同時檢查 覆寫參數是否會改變結果。

import unittest

from examples.<模組名稱> import MainTask


class MainTaskTests(unittest.TestCase):
    def test_run(self):
        self.assertEqual(MainTask(start=5).run()["result"], "value is 10")

    def test_parameter_override(self):
        self.assertEqual(MainTask(start=5, option=False).run()["result"], "value is 5")


if __name__ == "__main__":
    unittest.main()

然後使用 run-process 技能驗證。完整的運作範例位於 process_package/examples/word_report.py。