概要
Digdag はバッチ処理のワークフローを定義して定期的に実行するなどのタスク管理ツール1．
公式ドキュメント：
具体的な処理はシェルコマンドや Python や Ruby などで記述されたスクリプトを呼び出すことで行うが，本記事では Python ライブラリとしての digdag の型付けについてまとめる．
digdag の使い方
Digdag（アプリケーション）で Python を使用する際の基本的な処理は以下の2つ．
- Digdag からスクリプトを呼び出す
- Python関数の引数は，
*.digファイル内で定義された変数や組み込み変数で同名のものが自動的に渡される
- Python関数の引数は，
- Python からデータを渡す
-
digdag.env.store({'key': value})で辞書に値を追加する
-
他に子タスクを生成する機能などがある（参考: https://docs.digdag.io/python_api.html ）2．
特に後者では import digdag として digdag モジュールを参照するのだが，linterが効かないので設定を行う．
digdag モジュールの型ヒント
digdag モジュールはパッケージマネージャからインストールするのではなく、Digdag アプリから呼び出されたときに限り動的にモジュールがランタイム環境に生やされるという 邪悪な 実装がされているので静的解析が効かない．
このため，digdag モジュールの型付けを行うにはスタブファイル（ロジックを省略して型情報のみを記載したファイル）を作成する必要がある．
ソースコードを確認すると以下のようなクラス構造になっていることがわかるので，この通りにスタブを作成することで型チェックを通せるようになる．
from collections import OrderedDict
from typing import Any, Dict
class Env:
params: Any
subtask_config: OrderedDict
export_params: Dict[Any, Any]
store_params: Dict[Any, Any]
state_params: Dict[Any, Any]
subtask_index: int
def __init__(self, digdag_env_mod: Any) -> None:
...
def add_subtask(self, function: Any = None, **params: Any) -> None:
...
def export(self, params: Dict[Any, Any] = {}, **kwds: Any) -> None:
...
def set_state(self, params: Dict[Any, Any] = {}, **kwds: Any) -> None:
...
def store(self, params: Dict[Any, Any] = {}, **kwds: Any) -> None:
...
env: Env
上記は実装コードに stubgen を適用し，適当に整えたものである．
ここではパラメーターの型に制限がなく Any となっているが，実際にはデータのやり取りをJSON化して行うため，辞書はキーが str であるなどの制約をつけたほうがより正確な型チェックを行える．
from collections import OrderedDict
from typing import Any, Dict, TypeAlias
DictStrToAny: TypeAlias = Dict[str, Any]
class Env:
params: Any
subtask_config: OrderedDict
export_params: DictStrToAny
store_params: DictStrToAny
state_params: DictStrToAny
subtask_index: int
def __init__(self, digdag_env_mod: Any) -> None:
...
def add_subtask(self, function: Any = None, **params: Any) -> None:
...
def export(self, params: DictStrToAny = {}, **kwds: Any) -> None:
...
def set_state(self, params: DictStrToAny = {}, **kwds: Any) -> None:
...
def store(self, params: DictStrToAny = {}, **kwds: Any) -> None:
...
env: Env
valueの方は入れ子の辞書となることもあり得，ややこしくなるためここでは Any のままとした．また，より受け入れを広くして collections.abc.Mapping を使ったほうが適当かもしれない．
このあたりは各自で調整していただきたい．
mypyの型チェック設定
mypyで使用するには，stubs/digdag.pyiに上記のスタブファイルを記述し，
[mypy]
mypy_path = ./stubs
などのようにして指定する．
2つ目に示した，パラメータを Dict[str, Any] で受け取るバージョンで型チェックが通過することを確認している．
VSCodeでの設定
環境設定でパスを通せば Pylance などからも認識され，digdag モジュールの型情報を認識できるようになる．（ただし，Python 環境にモジュールがインストールされているわけではないので，インポート解決はできないようである）
{
...,
"python.analysis.extraPaths": [
"./stubs"
]
}
スタブ作成前：未知の状態で、シンタックスハイライトも効いていない状態から
スタブ作成後：メソッドであることが認識され、インターフェースが表示されるようになった。
関連記事