Claude Code with Dagster: Pipelines That Reason About Data
Why Dagster without CLAUDE.md produces Airflow-style pipelines without the lineage
Dagster is the asset-centric data orchestrator that reshaped the 2024-2026 data engineering stack. The model is simple: you declare the assets your pipeline produces (a table, a model, a CSV, a feature matrix), Dagster figures out the dependency graph from the asset definitions, and the platform handles materialisation, partitions, lineage, observability, and replays automatically. The problem is that Claude Code's training data is dominated by Airflow, which is task-centric. Without explicit constraints, Claude generates Dagster code that uses the lower-level @op and @job decorators in a procedural style that mimics Airflow DAGs and discards almost everything that makes Dagster valuable. The pipeline runs. The output looks right. But the lineage view is empty, the asset catalogue is empty, the partition history is empty, and the platform's UI shows a black box where there should be a dependency graph.
The most common Claude defaults that defeat Dagster's value: writing pipelines as @op plus @job instead of @asset (so the platform cannot infer the dependency graph or display asset lineage), defining IO inline within each op (so there is no IO manager and the asset cannot be reloaded for downstream runs), omitting partitions on time-series assets (so backfills cannot run by date range and historical replays require code changes), confusing resources with assets (so database connections are recreated per asset call instead of pooled), and ignoring auto-materialize policies entirely (so the pipeline only runs on manual or scheduled triggers and lineage-aware reactive materialisation is unused).
This guide covers the CLAUDE.md configuration that locks Claude Code into Dagster's asset-centric model: pipelines defined as @asset graphs with explicit IO managers, partitions on time-series and tenant assets, resources for shared connections, sensors for event-driven triggers, and auto-materialize policies that let upstream changes propagate downstream automatically. If you are building an LLM application backed by a Dagster-driven feature pipeline, Claude Code with Qdrant covers the vector store that Dagster assets often write to. For trace correlation between pipeline runs and observability data, Claude Code with Grafana covers the metrics dashboards your Dagster runs feed.
The Dagster CLAUDE.md template
The CLAUDE.md at your project root is read at the start of every Claude Code session. For a Dagster project it needs to declare: the Dagster version, the project layout, the asset-first programming model, the IO manager contract, the partition strategy, the resource pattern, and the hard rules that block the mistakes Claude makes most often.
# Dagster rules
## Stack
- dagster ^1.9 (Python 3.10+)
- dagster-webserver for the UI
- dagster-postgres OR dagster-cloud for production storage
- DAGSTER_HOME env var pointing to instance config dir
- Code locations defined in workspace.yaml
## Project structure
- my_project/
__init__.py , Definitions object (one per code location)
assets/ , Asset modules grouped by domain
raw.py , Raw extraction assets
transformed.py , Transformation assets
published.py , Output assets (dashboards, exports)
resources.py , Resource definitions (DB, S3, API clients)
io_managers.py , Custom IO managers
partitions.py , Partition definitions (daily, monthly, dynamic)
sensors.py , Event-driven triggers
schedules.py , Time-based triggers
jobs.py , Selection-based jobs (rare, prefer asset selections)
## Asset-first programming (MANDATORY)
- Pipelines MUST be declared as @asset graphs, NOT @op + @job
- @asset functions return the asset value, Dagster handles persistence via IO manager
- Dependencies come from function arguments (deps=, ins=, or argument name matching)
- @op + @job are valid ONLY for non-asset-producing workflows (rare)
## IO managers (REQUIRED for production)
- Default IO manager: dagster's built-in for development (writes to local fs)
- Production: custom IO manager per output type (S3, BigQuery, Postgres, Parquet)
- io_manager_key in @asset metadata selects which IO manager handles persistence
- IO managers separate the "what" (asset definition) from the "where" (storage)
## Hard rules
- NEVER write pipelines as @op + @job when @asset is appropriate
- NEVER perform IO inside an @asset body, use the IO manager
- NEVER instantiate database connections per asset, use a resource
- NEVER omit partitions on time-series or tenant-scoped assets
- NEVER hardcode credentials in asset code, use resources with config
- ALWAYS attach metadata to asset outputs (row counts, schemas, freshness)
- ALWAYS declare deps explicitly when dependency is by side-effect not data
Three rules here prevent the majority of value-loss outcomes Claude generates without them.
The asset-first rule is the most impactful because it determines whether Dagster's lineage, asset catalogue, and replay features work at all. An @op + @job pipeline runs successfully but produces no asset lineage, no asset catalogue entries, and no per-asset partition history. The Dagster UI shows job runs but no assets. From the platform's perspective, the pipeline produced nothing. Every other Dagster feature, from auto-materialize to backfills to data freshness checks, depends on assets being declared as assets.
The IO manager rule prevents the "pipeline ran but downstream cannot find the output" failure. When an asset performs IO inside its body (e.g. writing a Parquet file to a hardcoded path), the location is invisible to Dagster. Downstream assets cannot consume the output without duplicating the path knowledge. Backfills cannot reuse partitioned outputs because Dagster does not know they exist. The IO manager pattern centralises persistence: the asset returns a Python object, the IO manager writes it to the configured location, and downstream assets receive the deserialised object as their input.
The partition rule prevents the backfill-impossible failure mode. Time-series assets without partitions can only materialise the entire history as one run. Backfilling a missing week requires either custom code or rerunning the entire history. Partitioned assets let you run any subset by date range, replay individual partitions on failure, and parallelise materialisation across partitions. Claude omits partitions by default because they require an additional declaration. CLAUDE.md needs to make them mandatory for time-series and tenant-scoped data.
Project layout and Definitions object
A clean Dagster project starts with the Definitions object that wires assets, resources, sensors, and schedules together. This is the single entry point Dagster reads to discover the project.
# my_project/__init__.py
from dagster import Definitions, EnvVar, load_assets_from_modules
from dagster_aws.s3 import S3Resource
from . import assets
from .io_managers import s3_parquet_io_manager
from .resources import postgres_resource, openai_resource
from .sensors import new_files_sensor
from .schedules import daily_refresh_schedule
all_assets = load_assets_from_modules([assets])
defs = Definitions(
assets=all_assets,
resources={
"io_manager": s3_parquet_io_manager.configured({
"s3_bucket": EnvVar("DAGSTER_S3_BUCKET"),
"s3_prefix": "warehouse",
}),
"s3": S3Resource(),
"postgres": postgres_resource,
"openai": openai_resource,
},
sensors=[new_files_sensor],
schedules=[daily_refresh_schedule],
)
The Definitions object is loaded by Dagster's CLI and webserver. The resources dict provides shared connections that any asset can request. The IO manager registered as "io_manager" is the default for all assets (overridable per-asset).
Add a project layout section to CLAUDE.md:
## Project layout
- ONE Definitions object per code location, at my_project/__init__.py
- load_assets_from_modules() to auto-discover assets from a module
- Resources dict provides shared connections (DB, S3, API clients)
- IO managers registered under "io_manager" key are the default
- Per-asset IO manager: io_manager_key=... in @asset decorator
- Sensors and schedules registered explicitly in the Definitions
Assets, not ops and jobs
The fundamental shift Claude needs to make when generating Dagster code is from "tasks that move data around" to "data that exists, plus the logic that produces it". A daily orders aggregation in the asset-first model:
# my_project/assets/transformed.py
from dagster import asset, AssetIn, MetadataValue, DailyPartitionsDefinition
import pandas as pd
daily_partitions = DailyPartitionsDefinition(start_date="2025-01-01")
@asset(
partitions_def=daily_partitions,
group_name="orders",
io_manager_key="io_manager",
metadata={"owner": "data-team@claudify.tech"},
description="Raw order events for the partition date.",
)
def raw_orders(context) -> pd.DataFrame:
partition_date = context.partition_key
# Fetch the day's orders from the source DB
query = f"SELECT * FROM orders WHERE DATE(created_at) = '{partition_date}'"
df = context.resources.postgres.read_sql(query)
context.add_output_metadata({
"row_count": MetadataValue.int(len(df)),
"preview": MetadataValue.md(df.head().to_markdown()),
})
return df
@asset(
partitions_def=daily_partitions,
group_name="orders",
io_manager_key="io_manager",
ins={"raw_orders": AssetIn()},
description="Cleaned and validated orders, with derived columns.",
)
def cleaned_orders(context, raw_orders: pd.DataFrame) -> pd.DataFrame:
df = raw_orders.copy()
df = df.dropna(subset=["customer_id", "amount"])
df["amount_gbp"] = df["amount"] * df["currency_rate"]
df["created_date"] = pd.to_datetime(df["created_at"]).dt.date
context.add_output_metadata({
"row_count": MetadataValue.int(len(df)),
"dropped_rows": MetadataValue.int(len(raw_orders) - len(df)),
})
return df
@asset(
partitions_def=daily_partitions,
group_name="orders",
io_manager_key="io_manager",
ins={"cleaned_orders": AssetIn()},
description="Daily order metrics aggregated by customer segment.",
)
def daily_order_metrics(context, cleaned_orders: pd.DataFrame) -> pd.DataFrame:
agg = cleaned_orders.groupby("customer_segment").agg(
order_count=("order_id", "count"),
revenue_gbp=("amount_gbp", "sum"),
avg_order_value_gbp=("amount_gbp", "mean"),
).reset_index()
agg["partition_date"] = context.partition_key
context.add_output_metadata({
"segments": MetadataValue.int(len(agg)),
"total_revenue_gbp": MetadataValue.float(agg["revenue_gbp"].sum()),
})
return agg
What the asset-first model gives you that ops-and-jobs does not:
- Lineage: Dagster sees
raw_orders -> cleaned_orders -> daily_order_metricsand renders it as a graph. The lineage page shows the dependency, the materialisation history, and the metadata for every run. - Asset catalogue: each asset appears in the UI with its description, group, owner metadata, last materialisation time, and freshness. New team members can browse the catalogue to understand what data exists.
- Per-partition history: each daily partition is tracked independently. The UI shows which dates have been materialised, which have failed, and which are stale.
- Partition-aware backfills: backfilling missing dates is a UI operation. Select the asset, select the date range, click backfill. No code change.
- Reactive materialisation: an auto-materialize policy on
daily_order_metricscan trigger automatically whenevercleaned_ordersis materialised, no schedule required.
None of this works if the same pipeline is written as @op + @job. The pipeline runs, but the platform's value evaporates.
Add an assets section to CLAUDE.md:
## Assets
- @asset for any function that produces a stored data artifact
- partitions_def: required for time-series, tenant-scoped, or otherwise sliced data
- group_name: required, groups related assets in the UI for navigation
- io_manager_key: maps to a resource that handles persistence
- ins={...} for explicit dependency naming when argument names do not match
- metadata: dictionary of static metadata (owner, schema, source URL)
- description: required, surfaces in the asset catalogue
- add_output_metadata: per-materialisation runtime metadata (counts, samples)
IO managers
The IO manager is the contract for how assets are persisted and reloaded. The asset returns a Python object. The IO manager decides where it lives. Downstream assets receive the deserialised object as their argument.
A custom IO manager that writes Parquet to S3:
# my_project/io_managers.py
from dagster import (
ConfigurableIOManager,
InputContext,
OutputContext,
io_manager,
)
from dagster_aws.s3 import S3Resource
import pandas as pd
import io
class S3ParquetIOManager(ConfigurableIOManager):
s3_bucket: str
s3_prefix: str
s3: S3Resource
def _get_key(self, context):
# Build a hierarchical key from asset and partition
asset_path = "/".join(context.asset_key.path)
if context.has_partition_key:
return f"{self.s3_prefix}/{asset_path}/{context.partition_key}.parquet"
return f"{self.s3_prefix}/{asset_path}.parquet"
def handle_output(self, context: OutputContext, obj: pd.DataFrame) -> None:
buffer = io.BytesIO()
obj.to_parquet(buffer, compression="snappy")
buffer.seek(0)
key = self._get_key(context)
self.s3.get_client().put_object(
Bucket=self.s3_bucket,
Key=key,
Body=buffer.getvalue(),
)
context.add_output_metadata({
"s3_uri": f"s3://{self.s3_bucket}/{key}",
"size_bytes": buffer.getbuffer().nbytes,
})
def load_input(self, context: InputContext) -> pd.DataFrame:
key = self._get_key(context.upstream_output)
obj = self.s3.get_client().get_object(
Bucket=self.s3_bucket,
Key=key,
)
return pd.read_parquet(io.BytesIO(obj["Body"].read()))
s3_parquet_io_manager = S3ParquetIOManager(
s3_bucket="placeholder-bucket",
s3_prefix="placeholder-prefix",
s3=S3Resource(),
)
Three things the IO manager handles that asset code does not need to:
- Path generation: the key is derived from the asset name and partition key. Renaming an asset moves its data automatically.
- Serialisation: the handle_output method converts the Python object to bytes. Different IO managers can use different formats (Parquet, CSV, JSON, pickle).
- Reload: load_input deserialises when a downstream asset needs the upstream output. The downstream asset receives the DataFrame, not a path.
The pattern composes. A single project can have multiple IO managers (S3 Parquet for warehouse data, Postgres for OLTP-shaped data, local pickle for development). Each asset picks one via io_manager_key.
Add an IO managers section to CLAUDE.md:
## IO managers
- Subclass ConfigurableIOManager, implement handle_output and load_input
- _get_key pattern: derive storage location from asset_key and partition_key
- Register IO managers in the Definitions resources dict
- Default IO manager: "io_manager" key (used when @asset omits io_manager_key)
- Per-asset override: io_manager_key="my_custom_manager" in @asset
- Always add_output_metadata with the storage URI for traceability
- Test IO managers with a local in-memory implementation for unit tests
Partitions
Partitions slice an asset into independently materialisable units. The two most common patterns are daily partitions for time-series data and static partitions for tenant-scoped data.
# my_project/partitions.py
from dagster import (
DailyPartitionsDefinition,
HourlyPartitionsDefinition,
StaticPartitionsDefinition,
MultiPartitionsDefinition,
)
daily = DailyPartitionsDefinition(start_date="2025-01-01")
hourly = HourlyPartitionsDefinition(start_date="2025-01-01-00:00")
tenants = StaticPartitionsDefinition(["acme", "beta", "gamma"])
# Multi-partition: each combination of date and tenant is a unique partition
daily_by_tenant = MultiPartitionsDefinition({
"date": daily,
"tenant": tenants,
})
Using partitions in an asset:
from dagster import asset
from .partitions import daily_by_tenant
@asset(partitions_def=daily_by_tenant)
def tenant_daily_metrics(context):
partition = context.partition_key.keys_by_dimension
date = partition["date"]
tenant = partition["tenant"]
# fetch and aggregate for the (date, tenant) combination
The Dagster UI shows partitioned assets as a grid: rows per tenant, columns per date, cells showing the materialisation status (green = success, red = failed, grey = not yet run). Backfills select a rectangle of the grid and run all cells in parallel.
Dynamic partitions are added at runtime (e.g. new tenants onboarding):
from dagster import DynamicPartitionsDefinition
tenants = DynamicPartitionsDefinition(name="tenants")
# Add a new tenant from a sensor or schedule
context.instance.add_dynamic_partitions("tenants", ["new_customer"])
Add a partitions section to CLAUDE.md:
## Partitions
- DailyPartitionsDefinition for time-series (most common)
- HourlyPartitionsDefinition for high-frequency streams
- StaticPartitionsDefinition for fixed enumerable sets (tenants, regions)
- DynamicPartitionsDefinition for runtime-added partitions (new customers)
- MultiPartitionsDefinition for combinations (date x tenant)
- partitions_def on @asset is REQUIRED for any time-series or tenant-scoped data
- context.partition_key gives the current partition string in the asset body
- context.partition_key.keys_by_dimension for multi-partition combinations
Resources
Resources are shared, configurable connections that multiple assets use. The classic resource is a database connection pool, an S3 client, or an API client.
# my_project/resources.py
from dagster import ConfigurableResource, EnvVar
import psycopg2
from openai import OpenAI
class PostgresResource(ConfigurableResource):
connection_string: str
def read_sql(self, query: str):
import pandas as pd
return pd.read_sql(query, self.connection_string)
def execute(self, query: str):
with psycopg2.connect(self.connection_string) as conn:
with conn.cursor() as cur:
cur.execute(query)
class OpenAIResource(ConfigurableResource):
api_key: str
def get_client(self) -> OpenAI:
return OpenAI(api_key=self.api_key)
postgres_resource = PostgresResource(
connection_string=EnvVar("POSTGRES_CONNECTION_STRING"),
)
openai_resource = OpenAIResource(
api_key=EnvVar("OPENAI_API_KEY"),
)
Using resources in assets:
@asset
def enriched_users(context, postgres: PostgresResource, openai: OpenAIResource) -> pd.DataFrame:
df = postgres.read_sql("SELECT * FROM users")
client = openai.get_client()
# Use client for enrichment
return df
The resource argument name (postgres, openai) matches the key in the Definitions resources dict. Dagster injects the resource automatically. No global state, no per-asset connection instantiation.
Add a resources section to CLAUDE.md:
## Resources
- Subclass ConfigurableResource for stateful connections
- Register in Definitions resources dict
- EnvVar("...") for secrets and environment-specific config
- Asset signature: def my_asset(context, resource_name: ResourceType)
- One resource definition per external system (DB, S3, API)
- Resources are pooled across assets in the same run
- NEVER instantiate connections inside asset bodies, ALWAYS via resources
Sensors and schedules
Schedules trigger materialisation on a cron-like cadence. Sensors trigger on external events.
A daily schedule that materialises the order metrics pipeline:
# my_project/schedules.py
from dagster import ScheduleDefinition, AssetSelection, define_asset_job
daily_orders_job = define_asset_job(
name="daily_orders_job",
selection=AssetSelection.groups("orders"),
)
daily_refresh_schedule = ScheduleDefinition(
job=daily_orders_job,
cron_schedule="0 6 * * *",
execution_timezone="Europe/London",
)
A sensor that triggers when new files appear in S3:
# my_project/sensors.py
from dagster import sensor, SensorEvaluationContext, RunRequest, SkipReason
from dagster_aws.s3 import S3Resource
@sensor(job=daily_orders_job)
def new_files_sensor(context: SensorEvaluationContext, s3: S3Resource):
last_processed_key = context.cursor or ""
new_keys = []
paginator = s3.get_client().get_paginator("list_objects_v2")
for page in paginator.paginate(Bucket="incoming-data"):
for obj in page.get("Contents", []):
if obj["Key"] > last_processed_key:
new_keys.append(obj["Key"])
if not new_keys:
return SkipReason("No new files since last run.")
context.update_cursor(max(new_keys))
return RunRequest(
run_key=max(new_keys),
tags={"source_key": max(new_keys)},
)
The sensor's cursor pattern is critical. The cursor tracks what has been processed and survives across sensor evaluations. Without it, the sensor fires on every evaluation regardless of whether new work exists.
Add a sensors and schedules section to CLAUDE.md:
## Sensors and schedules
- Schedules: define_asset_job + ScheduleDefinition for cron triggers
- Sensors: @sensor for event-driven triggers
- AssetSelection.groups("X") to select all assets in a group
- AssetSelection.assets("a", "b") for specific assets
- AssetSelection.downstream("a") for an asset plus everything that depends on it
- Sensor cursor pattern: store last-processed marker, return SkipReason if no work
- execution_timezone REQUIRED on schedules (UTC default is rarely correct)
Auto-materialize policies
Auto-materialize policies let assets refresh themselves reactively when upstream changes. This eliminates many schedules and sensors entirely.
from dagster import AutoMaterializePolicy
@asset(
auto_materialize_policy=AutoMaterializePolicy.eager(),
)
def downstream_summary(upstream_data: pd.DataFrame) -> pd.DataFrame:
return upstream_data.groupby("category").sum()
AutoMaterializePolicy.eager() materialises the asset whenever its parents change. AutoMaterializePolicy.lazy() materialises only when downstream demand exists (e.g. the asset is referenced by a job run). Most analytics pipelines use eager for the top-of-funnel transformations and lazy for the leaves.
Add an auto-materialize section to CLAUDE.md:
## Auto-materialize
- AutoMaterializePolicy.eager() for assets that should react to upstream changes
- AutoMaterializePolicy.lazy() for assets that only run when needed
- Reduces schedule and sensor count significantly
- Lineage-aware: only re-materialises when parents actually change
- Combined with partitions: eager policy can backfill missing partitions automatically
Common Claude Code mistakes with Dagster
Six patterns Claude generates incorrectly without CLAUDE.md constraints, with the correct replacement for each.
1. Ops and jobs instead of assets
Claude generates: @op functions composed into a @job, mimicking Airflow DAG syntax.
Correct pattern: @asset functions with dependency inferred from arguments.
2. IO inside asset bodies
Claude generates: df.to_parquet("/tmp/output.parquet") inside an @asset body.
Correct pattern: asset returns the DataFrame, IO manager handles persistence to S3 or warehouse.
3. Missing partitions
Claude generates: @asset for a daily-refreshing table with no partitions_def.
Correct pattern: @asset(partitions_def=daily_partitions) so backfills and per-date history work.
4. Connections in asset bodies
Claude generates: conn = psycopg2.connect(...) inside the @asset body.
Correct pattern: def my_asset(context, postgres: PostgresResource) with the connection injected.
5. No metadata
Claude generates: @asset functions that return data with no add_output_metadata call.
Correct pattern: every asset records row counts, sample previews, and source URIs as metadata.
6. Hardcoded execution timezone
Claude generates: ScheduleDefinition(cron_schedule="0 6 * * *") with no execution_timezone.
Correct pattern: execution_timezone="Europe/London" (or the project's canonical timezone), never default UTC silently.
Add a common mistakes section to CLAUDE.md with these six pairs.
Permission hooks for Dagster scripts
A Dagster project accumulates scripts: backfill runners, asset-state inspectors, migration helpers, materialisation triggers. Some are read-only. Some mutate production state. Permission hooks gate the destructive ones.
In .claude/settings.local.json:
{
"permissions": {
"allow": [
"Bash(dagster asset list*)",
"Bash(dagster asset materialize --dry-run*)",
"Bash(dagster job list*)",
"Bash(python scripts/inspect_assets.py*)"
],
"deny": [
"Bash(dagster asset materialize *)",
"Bash(dagster job execute*)",
"Bash(dagster asset wipe*)",
"Bash(python scripts/backfill.py*)"
]
}
}
Listing assets and dry-running materialisations are safe. Wiping assets, executing jobs, or running backfills can mutate production data warehouses. The deny list forces Claude to surface those operations as prompts rather than running them as part of an automated workflow.
Building Dagster pipelines that surface lineage
The Dagster CLAUDE.md in this guide produces data pipelines where every step is an @asset with explicit lineage, every output goes through an IO manager that knows where to store and how to reload it, partitions enable per-date backfills and replay, resources provide pooled connections, and sensors plus auto-materialize policies handle event-driven and reactive materialisation.
The underlying principle is the same as any orchestrator integration with Claude Code. Dagster without a CLAUDE.md produces pipelines that run successfully but discard everything the platform exists to provide: the asset catalogue is empty, the lineage graph is empty, the partition history is empty, and the rich metadata that makes the UI valuable is absent. Claude defaults to the Airflow-shaped patterns that dominate its training data, and Dagster's value evaporates. The CLAUDE.md template removes those defaults by making the asset-centric pattern the only pattern Claude can generate.
For the next layer of data infrastructure, Dagster works alongside Claude Code with Qdrant for vector pipelines that write to vector databases, and Claudify includes a Dagster-specific CLAUDE.md template with the asset-first pattern, IO manager contract, partition strategy, resource pooling, auto-materialize policies, and all six common-mistake rules pre-configured.
Get Claudify. Ship Dagster pipelines that actually use the platform you chose.
More like this
Ready to upgrade your Claude Code setup?
Get Claudify
