Claude CodeでAirflowのDAGを書く — dags testとimport errorの確認手順
Claude CodeにAirflowのDAGを書かせたら、import errorの確認とdags testの実行までを検証コマンドとして固定する。CLAUDE.mdとhooksの設定例つき。
AirflowのDAGをClaude Codeに書かせるときの検証は3段で足りる
AirflowのDAGは、書いた時点では動くかどうかが分かりません。Schedulerに読み込まれて初めて、importの失敗や依存関係の不整合が表に出ます。Claude Codeに任せるなら、書かせる前に「何を実行すれば合格か」を決めておくのが近道です。
Claude Codeの公式ガイドも、テストやビルドの終了コード、リンターなど、Claudeが会話の中で読める確認手段を渡すよう勧めています。AirflowにはそのままCLIとして使える確認手段が3つあります。
| 段 | コマンド | 見つけるもの |
|---|---|---|
| 1 | コマンドpython dags/xxx.py | 見つけるもの構文エラー、未インストールの依存 |
| 2 | コマンドairflow dags list-import-errors -l | 見つけるものAirflowが読み込む段階のimport error |
| 3 | コマンドairflow dags test <dag_id> | 見つけるもの1回分のDagRunを通しで動かしたときの失敗 |
この記事では、この3段をClaude Codeの検証ループに組み込む手順を、CLAUDE.mdの記述、pytest、hooksの設定例まで順に説明します。Airflowの稼働状況を監視する側の話はAirflow MCPサーバーでDAGの実行状況をClaudeに監視させるにあります。ここで扱うのは、DAGを実装する側の検証です。
第1段Python実行でDAGファイルのロードエラーを潰す
AirflowのBest Practicesは、DAGファイルをそのままPythonで実行する「Dag Loader Test」を紹介しています。エラーなく終われば、構文エラーや未インストールの依存が無いことが分かります。
python dags/daily_report.py注意点は実行環境です。Schedulerと同じ依存関係・環境変数・共通コードを持つ環境で読み込む必要があります。手元の仮想環境とScheduler側のイメージで入っているパッケージが違うと、ここで通っても本番で落ちます。Claude Codeに実行させるときは、どの仮想環境で走らせるかをCLAUDE.mdに書いておくと食い違いを避けられます。
第2段list-import-errorsでAirflowの読み込み結果を確認する
第1段が通っても、Airflow側の読み込みで失敗することがあります。CLIリファレンスでは、airflow dags list-import-errorsが「importエラーのあるDAGの一覧」を返すコマンドとして載っています。
airflow dags list-import-errors -l -o json-l(--local)は、ローカルで解析したDAGとそのimport errorを表示し、DBにシリアライズ済みの内容を無視します。書き換え直後の確認では、DBの古い状態に引きずられないこのオプションが向きます。-oにはtable、json、yaml、plainを指定できます。Claudeに読ませるなら、機械的に扱いやすいjsonが選択肢です。
DB側の内容が古いと疑うときは、airflow dags reserializeがあります。シリアライズされたDAGがAirflowのバージョンとずれたときに役立つコマンドです。
DAGが一覧に出ないときに疑う3点
DAGのimport errorが無いのに見つからない、というケースがあります。Loading Dagsの節から、Claudeが書いたコードで起きやすい3点を挙げます。
- 読み込まれるのはトップレベルのDAGオブジェクトだけです。関数の中でDAGを作っても、
globals()に無ければ登録されません。@dagで作った関数は、ファイル内で1回呼び出してトップレベルの変数に代入する必要があります。 - ファイルに「airflow」と「dag」の文字列が無いと探索対象になりません(dag_discovery_safe_mode)。最適化のための既定動作です。ラッパー経由でDAGを定義する構成では、
[core] dag_discovery_safe_modeをFalseにします。 .airflowignoreに一致したファイルも読み込まれません。Claudeが新しいサブディレクトリにDAGを置いたときは、無視パターンに当たっていないかを見ます。
第3段airflow dags testで1回分のDagRunを動かす
読み込みが通ったら、実際にタスクを動かします。airflow dags testは、指定したDAGの1回分のDagRunを、論理日付を指定して実行するコマンドです。
airflow dags test daily_report 2026-10-01AirflowのDagsドキュメントによれば、dag.test()はLocalExecutorに近い「模擬的な実行フロー」でDAGの中身を動かします。CLIのリファレンスでは、既定ではExecutorを使わず、--use-executorを付けると環境に設定されたExecutorで動かすと説明されています。
使うオプションを絞ると、Claudeに任せる指示も短くなります。
| オプション | 役割 |
|---|---|
-f(--dagfile-path) | 役割DAGファイルを直接指定する |
-B(--bundle-name) | 役割複数のDAGバンドルがあるときに指定する |
-c(--conf) | 役割DagRunのconfにJSON文字列を渡す |
--mark-success-pattern | 役割正規表現に一致するtask_idを実行せず成功扱いにする |
--save-dagrun | 役割実行後の図をファイルに保存する(拡張子で形式が決まる) |
--mark-success-patternは、センサーや後始末のタスクを局所テストで飛ばす用途だと、CLIリファレンスに書かれています。外部システムの到着を待つセンサーを含むDAGなら、次のように飛ばせます。
airflow dags test daily_report 2026-10-01 \
--mark-success-pattern "wait_for_.*"DAGファイルの末尾に次の2行を置く方法もあります。同じドキュメントが「DAGの妥当性を確かめる最も簡単な方法」として紹介している書き方です。
if __name__ == "__main__":
dag.test()こうすると、第1段のpython dags/daily_report.pyがそのまま模擬実行になります。AIRFLOW_HOMEなど、必要な環境が揃っていることが前提です。
タスクの中身は実際に動くため、データベースへの書き込みや外部APIの呼び出しを含むDAGでは、接続先を検証用の環境に向けてから実行させます。接続先の切り替えは、環境変数でAIRFLOW_CONN_<CONN_ID>を与える方法がBest Practicesにあります。
CLAUDE.mdに検証手順を書いてDAG作成を依頼する
3段のコマンドが決まったら、CLAUDE.mdに書き込みます。テストの実行方法や好みのテストランナーのように、コードから読み取れない情報はCLAUDE.mdに置くのが定石です。次の断片はその一例です。
## Airflow DAGの検証手順
DAGを追加・変更したら、完了を宣言する前に次を順に実行し、出力を貼ること。
1. `python dags/<file>.py` がエラー無しで終わる
2. `airflow dags list-import-errors -l -o json` に対象ファイルが出ない
3. `airflow dags test <dag_id> <論理日付>` が成功する
- センサーは `--mark-success-pattern` で飛ばす
- 外部への書き込みを行うタスクは、検証用の接続先でのみ実行する
実行環境は `.venv`(Scheduler と同じ requirements)を使う。依頼文は、書く内容と合格条件をセットにします。
dags/daily_report.py に、売上CSVを集計して日次レポートを作るDAGを追加して。
タスクは extract → transform → load の3つ。
CLAUDE.md の検証手順を全部通して、各コマンドの出力を見せてから完了と報告して。最後の一文が効きます。成功したと言わせるのではなく、テスト出力や実行したコマンドとその結果を見せる形にするとレビューが速いと述べています。
pytestのDagBagテストで読み込みエラーを継続的に検知する
CLIは手動の確認に向きますが、変更のたびに回すならpytestに載せます。Best Practicesには、DagBagを使った読み込みテストの例があります。
import pytest
from airflow.dag_processing.dagbag import DagBag
@pytest.fixture()
def dagbag():
return DagBag()
def test_dag_loaded(dagbag):
dag = dagbag.get_dag(dag_id="daily_report")
assert dagbag.import_errors == {}
assert dag is not None
assert len(dag.tasks) == 3公式の例ではdag_idがhello_world、タスク数が1ですが、ここでは書かせたDAGに合わせて置き換えています。importパスはAirflow 3系の書き方で、2.x系ではfrom airflow.models import DagBagです。hooksのスクリプトも同様に読み替えます。import_errors == {}の確認は、DAGが1本壊れたときにテスト全体が落ちるので、別のDAGを巻き込んだ破壊にも気づけます。
構造を固定したい場合は、Best Practicesに「コード生成されたDAGの構造を辞書と比べるテスト」の例もあります。task_idの集合と、各タスクのdownstream_task_idsを突き合わせる形です。Claudeに「依存関係の期待値を辞書で書いてから実装する」と頼むと、>>の向きを間違えたときに失敗として表に出ます。
VariableやConnectionに依存するコードは、環境変数で擬似的に用意できます。VariableにはAIRFLOW_VAR_{KEY}、ConnectionにはAIRFLOW_CONN_{CONN_ID}を使い、mock.patch.dict("os.environ", ...)で与える方法を紹介しています。DBへ保存する方法より速いと説明されています。
hooksでDAG編集のたびにimport errorを確認させる
CLAUDE.mdの指示は守られないことがあります。確実に回したいなら、hooksで機械的に走らせます。PostToolUseは、ツール呼び出しが成功した後のイベントです。matcherにEdit|Writeを指定すると、ファイル編集の直後だけに走ります。
PostToolUseはブロックできないイベントですが、終了コード2のときのstderrはClaudeに表示されます。つまり、失敗内容をClaudeに読ませて直させる用途に合います。
.claude/settings.jsonには次のように登録します。
{
"hooks": {
"PostToolUse": [
{
"matcher": "Edit|Write",
"hooks": [
{
"type": "command",
"command": "\"$CLAUDE_PROJECT_DIR\"/.claude/hooks/check-dag.sh"
}
]
}
]
}
}スクリプトは、編集されたファイルがdags/配下のPythonファイルのときだけ、DagBagのimport_errorsを確認します。次はこの記事で組んだ一例で、公式のサンプルをそのまま写したものではありません。
#!/bin/bash
# .claude/hooks/check-dag.sh
INPUT=$(cat)
FILE=$(echo "$INPUT" | jq -r '.tool_input.file_path // empty')
case "$FILE" in
*/dags/*.py) ;;
*) exit 0 ;;
esac
python - <<'PY' || exit 2
import sys
from airflow.dag_processing.dagbag import DagBag
errors = DagBag().import_errors
for path, err in errors.items():
print(f"{path}: {err}", file=sys.stderr)
sys.exit(1 if errors else 0)
PYDagBag()は設定済みのDAGフォルダを読むので、編集したファイル以外のエラーも拾います。DAGが多い環境では読み込みに時間がかかる点に注意します。
あわせて、確認コマンドの実行許可をpermissionsで事前に与えておくと、承認待ちで止まらなくなります。許可ルールはBash(airflow dags list-import-errors *)のように、サブコマンドまでを固定して*を後ろに置く形が公式の書き方です。airflow dags testはタスクを実行するため、許可に入れるか承認制のままにするかは、接続先の安全性で判断します。
Stopやテスト自動実行への組み込みは、hooksでテストを自動実行する方法で別の観点から比較しています。
Claudeの修正ループで起きやすいつまずき
検証コマンドを渡すと、Claudeはエラーを読んで直す反復に入ります。ただし、Airflow特有の癖で空回りすることがあります。
- トップレベルの重い処理は、DAGファイルのパースのたびに走ります。import文もトップレベルのコードに含まれるので、重いライブラリはタスクの関数内でimportします。
- Variableをトップレベルで読むと、パースのたびにネットワークとDBへのアクセスが飛びます。必要ならJinjaテンプレート経由で渡し、実行時まで読み込みを遅らせる方法があります。
- import errorを
try/exceptで包ませると、検証だけ通って本番で動かないDAGができます。依頼文に「原因を直す。エラーの抑制はしない」と添えておくと、Claudeのガイドが示す「根本原因を直す」型の依頼になります。 airflow dags testのlogical_dateは省略できます。ただ、日付に依存するテンプレートを使うDAGでは、明示した方が結果を再現できます。
まとめ
ClaudeにAirflowのDAGを書かせる鍵は、書かせる前に合格条件をコマンドで固定することです。Pythonでのロード、list-import-errors -l、dags testの3段をCLAUDE.mdに書き、変更のたびにpytestのDagBagテストで回せば、「動いたはず」の報告が実行結果に変わります。hooksを足すと、編集直後にimport errorがClaudeへ返るようになります。
同じ「宣言的に書いてテストで固める」流れは、Dagsterのアセット定義でも使えます。dbtのモデルを含むパイプラインなら、dbt MCPサーバーでDAGの実行とリネージを確認できます。