プロセスを作成する
すべてのパスはリポジトリルートからの相対パスです。パッケージは 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 をオーバーライドしたりしないでください。
子クラスのルール
from toy_process import Processでインポートし、サブクラス化します。set_parameters(self)はデフォルト値のdictを返します。ない場合は{}を使用します。set_steps(self)はlistを返します。各要素は次のいずれかです:- 束縛メソッド:
self.count_words、括弧なし、または - Process インスタンス:
CleanText(lowercase=self.params["lowercase"])。
- 束縛メソッド:
- 各ステップメソッドのシグネチャは
def name(self, context):です。- 設定は
self.params["..."]から読み取ります。 - データは
context["..."]で読み書きします。 - 何も返しません。戻り値は無視されます。
- 設定は
- サブプロセスクラスは、それを使用するクラスの上に定義するか、インポートします。
テンプレート
"""プロセスが何をするか1行で説明します。
process_package/ から実行:
python -m examples.<module_name>
"""
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"])
作成するファイル
process_package/examples/<module_name>.py:テンプレートからのプロセス。process_package/tests/test_<module_name>.py:.run()を呼び出し、返された context のキーに対してアサーションを行うunittestテスト。パラメータのオーバーライドで結果が変わることも確認します。
import unittest
from examples.<module_name> 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 にあります。