Claude CodeでDagsterのアセットを定義する — テストしやすい宣言的パイプラインの書き方

Dagsterのアセットとは

Dagsterのアセットとは、データパイプラインが生成する成果物(テーブルやファイルなど)をPythonの関数として宣言的に定義したものです。実行順序のタスクを直接書くのではなく、「このデータはどう作られるか」を関数単位で書くと、Dagsterが依存関係のグラフを自動で組み立てます。

@dg.asset デコレータを関数に付けるだけで1つのアセットが定義でき、deps 引数で上流のアセットを指定すれば依存関係も明示できます。最小構成は次のとおりです。

import dagster as dg
 
@dg.asset
def hello(context: dg.AssetExecutionContext):
    context.log.info("Hello!")
 
@dg.asset(deps=[hello])
def world(context: dg.AssetExecutionContext):
    context.log.info("World!")

Claude Codeにこの構文を渡すと、既存のPythonコードの型ヒントとdocstringの読解から自然に書き分けます。アセット定義そのもののコード生成は他のPythonコードと変わりませんが、Dagster特有のCLI操作とテスト設計にClaude Codeを使うところに実務的な価値があります。

Claude CodeでDagsterプロジェクトを立ち上げる手順

新規プロジェクトの作成からアセットのスキャフォールドまで、Dagster公式のCLIをClaude Codeに実行させれば一連の作業が数コマンドで終わります。前提となる環境はPython 3.10以上で、パッケージ管理にはuvかpipのどちらかを使います。以下はuvを使う手順です。

手順1: プロジェクトを作成する

uvx create-dagster@latest project dagster-quickstart
cd dagster-quickstart
source .venv/bin/activate
uv add pandas

create-dagster CLIはpyproject.tomlとsrc以下のパッケージ構成を自動生成し、プロンプトでyと答えるとuv syncまで実行します。pipを使う場合はcreate-dagster projectを実行してから仮想環境を手動で作成し、pip install --editable . --group devでプロジェクトを編集可能な状態にします。

手順2: アセットファイルをスキャフォールドする

dg scaffold defs dagster.asset assets.py

dg scaffold defsは実行したディレクトリに関わらず、生成物を必ず<project-name>/src/<project_name>/defs/配下に置きます。生成直後のファイルはコメントアウトされた雛形なので、ここからClaude Codeに実装を書かせます。

手順3: 定義を確認する

dg list defs

dg list defsはプロジェクト内のアセット・依存関係・種別を表形式で一覧します。Claude Codeが書いたコードがDagsterに正しく認識されているかを、実際に実行せずに確認できます。

アセットデコレータの使い分け

Dagsterは1つの関数から1つのアセットを作る@assetのほかに、用途別の3種類のデコレータを用意しています。Claude Codeに実装を依頼するときは、どのデコレータを選ぶかを先に決めておくと、生成されるコードの構造が安定します。

@dg.multi_assetはAssetSpecのリストで複数の出力を宣言し、MaterializeResultをyieldして各アセットの結果を返します。

import dagster as dg
 
@dg.multi_asset(specs=[dg.AssetSpec("asset_one"), dg.AssetSpec("asset_two")])
def my_multi_asset():
    yield dg.MaterializeResult(asset_key="asset_one", metadata={"num_rows": 10})
    yield dg.MaterializeResult(asset_key="asset_two", metadata={"num_rows": 24})

アセットにはcode_versionという引数を付けられます。コードが変わったのに再計算していないアセットをUI上で見分けられるようになり、Claude Codeにリファクタリングを依頼するときも変更箇所の追跡に使えます。関数のシグネチャからスキーマ相当の情報を読み取る発想は、MCP Python SDKのデコレータでツールを定義する実装パターンで扱うツール定義の設計とも近いものがあります。

アセットのキーとメタデータを設計する

アセットにはグループ名や所有者、複数階層のキーを付けられます。Claude Codeにプロジェクト全体の実装を任せるときは、これらの引数を先に決めておくと、後からアセットを探しやすくなります。

group_nameはDagster UIの画面でアセットをまとめる単位です。ownersは担当者やチームを示すメタデータで、障害発生時の連絡先として機能します。

import dagster as dg
 
@dg.asset(deps=[weekly_sales], group_name="sales")
def weekly_sales_report(
    context: dg.AssetExecutionContext,
) -> None:
    context.log.info("Loading data for my_dataset")

アセットがファイルシステムのように階層を持つ場合は、key_prefix引数で複数階層のキーを組み立てます。下流のアセットが上流のキーを参照するときはAssetInで同じkey_prefixを指定します。

import dagster as dg
 
@dg.asset(key_prefix=["one", "two", "three"])
def upstream_asset():
    return [1, 2, 3]
 
@dg.asset(ins={"upstream_asset": dg.AssetIn(key_prefix=["one", "two", "three"])})
def downstream_asset(upstream_asset):
    return upstream_asset + [4]

Claude Codeにこうしたメタデータ付けを依頼するときは、命名規則(グループ名の単位、キーの階層の切り方)を先にCLAUDE.mdなどへ書いておくと、プロジェクト全体で表記が揃います。

テストしやすさが宣言的アセット定義の強み

Dagsterのアセットは普通のPython関数なので、Dagsterのランタイムを起動せずに直接呼び出してテストできます。これが宣言的アセット定義の実務上の強みで、Claude Codeに実装させたコードをその場で単体テスト化できる理由でもあります。

引数を取らないアセットは、そのまま呼び出すだけでテストが書けます。

# tests/test_assets.py
def test_loaded_file() -> None:
    assert loaded_file() == "contents"

context引数を使うアセットはdg.build_asset_context()でモックのコンテキストを組み立て、パーティションキーなどを注入します。

def test_loaded_file() -> None:
    context = dg.build_asset_context(partition_key="2024-08-16")
    assert loaded_file(context) == "Contents for August 16th, 2024"

外部サービスに依存するアセットは、リソースをモックに差し替えて呼び出します。

from unittest import mock
 
def test_file() -> None:
    mocked_resource = mock.Mock(spec=S3FileManager)
    mocked_resource.read_data.return_value = "contents"
    assert loaded_file(mocked_resource) == "contents"

Claude Codeに「このアセットの単体テストを書いて」と依頼すると、アセットの引数構成(config・resources・context)を読み取って、上のいずれかのパターンに沿ったテストコードを提案します。ただし@multi_assetで上流に依存があるケースだけは例外です。input_valuesが使えないため、モックのIOマネージャーを用意して上流のテストデータを返す実装が必要になります。この制約をプロンプトに書き添えておかないと、動かないテストコードが出てくることがあります。

アセットの実行結果そのものの品質を検証したい場合は、単体テストとは別にアセットチェックという仕組みも用意されています。単体テストが関数の計算ロジックそのものを確認するのに対して、アセットチェックは実際にマテリアライズされたデータが期待どおりの中身になっているかをランタイムで検証します。行数がゼロでないか、欠損値の割合が閾値を超えていないか、といったデータ品質の観点はアセットチェック側で書き、ロジックの正しさは単体テスト側で書く、という役割分担をClaude Codeへのプロンプトに含めておくと、生成されるテストコードの粒度が安定します。

dagster-expertスキルとDagster+ MCPサーバーを併用する

Dagsterは、Claude Code向けの公式プラグインをGitHub上で配布しています。dagster-expertスキルをインストールすると、アセットパターンやCLIコマンドの知識をClaude Codeに直接持たせられます。

/plugin marketplace add dagster-io/skills
/plugin install dagster@dagster
/dagster-expert "アセットのスキャフォールドの仕方を教えて"

インストール後は/pluginでInstalledタブを開き、dagster: enabledになっているかを確認します。プラグイン名はdagsterで、呼び出すスラッシュコマンドが/dagster-expertという別名になっている点に注意します(dagster-expertという名前のプラグインはすでに廃止され、Dagsterの知識を持たないスタブに置き換わっています)。

dagster-expertスキルがコード生成を助けるのに対し、Dagster+ MCPサーバーはデプロイ済みのDagster+環境を直接操作します。実行済みRunのログ取得、アセットのマテリアライズ、アラートポリシーの作成・更新まで、Claude Codeからひと続きの会話で行えます。

claude mcp add --transport http dagster-plus https://mcp.agent.dagster.cloud/mcp

接続後は/mcpでdagster-plusを選び、ブラウザ認証を済ませれば使えるようになります。EUリージョンの組織はDAGSTER_CLOUD_MCP_URL環境変数でエンドポイントを切り替えます。MCPサーバーが公開するツールの数が増えるほどコンテキストを消費する点は、MCPのツール定義はなぜコンテキストを圧迫するのかで扱った制約と同じ理屈です。Dagster+ MCPは実行系の操作に絞って導入し、コード生成はdagster-expertスキルに任せる、という役割分担が現実的です。

よくあるつまずき

• 仮想環境を有効化し忘れる: dg scaffoldやdg list defsはプロジェクトの仮想環境内でしか動きません。source .venv/bin/activateを忘れるとdgコマンド自体が見つからずに失敗します。
• multi_assetの依存を単純にテストしようとする: @multi_assetで上流アセットに依存する構成はinput_valuesが使えず、モックのIOマネージャーを用意しないとテストが組めません。
• code_versionを設定せずに再計算を追う: code_versionを付けていないアセットは、UI上でコード変更の有無を見分けられません。Claude Codeにリファクタリングを任せる前に、変更を追跡したいアセットへ付けておくと差分が見えます。
• dagster-expertとDagster+ MCPの役割を混同する: 前者はコード生成の支援、後者はデプロイ済み環境への操作です。ローカル開発だけならdagster-expertのインストールで十分で、MCPサーバーの接続は不要です。

まとめ

Claude CodeでDagsterのアセットを書く作業は、コード生成そのものより、CLIの実行とテスト設計への使い方で差が出ます。dg scaffold defsでひな形を作り、@dg.asset系のデコレータで依存関係を宣言し、直接呼び出しで単体テストを書く、という一連の流れをClaude Codeに任せられます。公式のdagster-expertスキルを導入すれば、CLIコマンドやアセットパターンの知識をコンテキストに持たせたまま作業でき、Dagster+ MCPサーバーはデプロイ後の運用操作を担います。まずはローカルプロジェクトでアセットとテストを一往復させ、必要に応じてMCPサーバーを追加する順番が扱いやすいところです。