Dagster Pipelines skill for Codex

Done well means: the asset graph mirrors the real data products, every asset materialization is reproducible for its partition, external systems sit behind resources configured from the environment, and checks block bad data from flowing downstream.

Before writing code

1. Read AGENTS.md, pyproject.toml (look for [tool.dagster] or [tool.dg]), and the definitions module. dagster --version; this skill targets Dagster 1.9+ with the import dagster as dg style.
2. Find how the project is laid out: one Definitions object, or the newer dg project layout with a defs/ folder loaded automatically. Follow it.
3. Load the code once before changing it: dg check defs (dg projects) or dagster definitions validate -m <module> and fix existing errors first.

Assets first

Model the things that persist (tables, files, ML models) as assets, not the steps that build them. Ops and graphs remain for imperative tasks that produce nothing durable.

import dagster as dg
from .resources import WarehouseResource

daily = dg.DailyPartitionsDefinition(start_date="2025-01-01", timezone="UTC")

@dg.asset(
    partitions_def=daily,
    group_name="orders",
    kinds={"snowflake"},
    owners=["team:data-platform"],
)
def raw_orders(context: dg.AssetExecutionContext, warehouse: WarehouseResource) -> dg.MaterializeResult:
    day = context.partition_key                      # "2025-09-01"
    rows = warehouse.replace_partition("raw.orders", day, fetch_orders(day))   # idempotent
    return dg.MaterializeResult(metadata={"rows": rows, "partition": day})

@dg.asset(partitions_def=daily, deps=[raw_orders], group_name="orders")
def daily_revenue(context: dg.AssetExecutionContext, warehouse: WarehouseResource) -> None:
    warehouse.execute(
        "delete from marts.daily_revenue where day = %(d)s; "
        "insert into marts.daily_revenue select %(d)s, sum(amount) from raw.orders where day = %(d)s",
        {"d": context.partition_key},
    )

Dependency style	Use when
deps=[upstream]	Data lives in a warehouse or lake; the asset reads it itself (recommended default)
Function argument def b(a: pd.DataFrame)	Small in-memory handoff through an IO manager

The default IO manager pickles to local disk. That is fine for dagster dev, wrong for production. Either use deps= with explicit storage, or configure a real IO manager.

Resources

# resources.py
import dagster as dg

class WarehouseResource(dg.ConfigurableResource):
    dsn: str
    role: str = "transformer"

    def execute(self, sql: str, params: dict | None = None) -> None: ...
    def replace_partition(self, table: str, day: str, rows) -> int: ...

defs = dg.Definitions(
    assets=[raw_orders, daily_revenue],
    resources={"warehouse": WarehouseResource(dsn=dg.EnvVar("WAREHOUSE_DSN"))},
)

dg.EnvVar is resolved at run time and never shown in the UI. Never pass secrets as literal strings or os.environ[...] at import time.

Running things on time

Need	Use
Fixed cron for a set of assets	dg.define_asset_job + dg.ScheduleDefinition(job=..., cron_schedule=..., execution_timezone="UTC")
Partitioned job, one run per new partition	dg.build_schedule_from_partitioned_job(job)
React to files, API state or external events	@dg.sensor with a cursor and run_key
“Update when parents update” or “keep fresh by cron”	Declarative automation: automation_condition=dg.AutomationCondition.eager() or .on_cron("0 * * * *")
Another asset materialized	Asset sensor or automation conditions, not polling

orders_job = dg.define_asset_job("orders_job", selection=dg.AssetSelection.groups("orders"),
                                 partitions_def=daily)
orders_schedule = dg.build_schedule_from_partitioned_job(orders_job, hour_of_day=3)

@dg.sensor(job=orders_job, minimum_interval_seconds=120)
def late_files(context: dg.SensorEvaluationContext):
    seen = context.cursor or ""
    new = sorted(f for f in list_landing_files() if f > seen)
    if not new:
        return dg.SkipReason("no new files")
    for f in new:
        yield dg.RunRequest(run_key=f, partition_key=partition_from_filename(f))
    context.update_cursor(new[-1])

run_key makes the sensor idempotent: the same key never launches twice. Automation conditions only run when the automation condition sensor is turned on for the deployment; check that in the UI before relying on them.

Asset checks

@dg.asset_check(asset=raw_orders, blocking=True)
def raw_orders_has_rows(warehouse: WarehouseResource) -> dg.AssetCheckResult:
    n = warehouse.scalar("select count(*) from raw.orders where day = current_date - 1")
    return dg.AssetCheckResult(passed=n > 0, metadata={"rows": n})

blocking=True stops downstream assets in the same run when the check fails. Put checks on assets other teams consume; leave deep rule suites to $data-quality-checks.

dbt integration

from dagster_dbt import DbtCliResource, dbt_assets

@dbt_assets(manifest=dbt_project_dir / "target" / "manifest.json")
def dbt_models(context: dg.AssetExecutionContext, dbt: DbtCliResource):
    yield from dbt.cli(["build"], context=context).stream()

Each dbt model becomes an asset with lineage to upstream Dagster assets via dbt sources. The manifest must be generated (dbt parse) at build or deploy time.

Backfills

Launch partition backfills from the UI (select the asset, Materialize, choose a range) or the CLI for single partitions: dagster asset materialize -m my_project.definitions --select daily_revenue --partition 2025-09-01. Before a large backfill, confirm with the user: partition range, run count, and concurrency. Limit concurrency with run queue settings or pools so the warehouse is not flooded.

Testing

import dagster as dg
from my_project.assets import raw_orders, daily_revenue

def test_orders_pipeline():
    result = dg.materialize(
        [raw_orders, daily_revenue],
        partition_key="2025-09-01",
        resources={"warehouse": FakeWarehouse()},
    )
    assert result.success
    mats = result.asset_materializations_for_node("raw_orders")
    assert mats[0].metadata["rows"].value > 0

def test_definitions_load():
    from my_project.definitions import defs
    dg.Definitions.validate_loadable(defs)

Assets are plain functions: call them directly with dg.build_asset_context(partition_key=...) for focused unit tests.

Gotchas

• Heavy work at import time (API calls, reading manifests from the network) makes the code location fail to load or load slowly. Keep imports cheap.
• Upstream and downstream partition definitions must line up, or need an explicit partition_mapping; otherwise dependencies resolve to the wrong day.
• Changing an asset key orphans its history. Rename with care and tell the user.
• A sensor without run_key or cursor relaunches the same work on every tick.
• dagster dev uses a temporary instance unless DAGSTER_HOME is set; run history disappears.

Verification and report

dg check defs            # or: dagster definitions validate -m my_project.definitions
pytest tests/ -q
dagster dev              # materialize one partition in the UI and inspect metadata

**Assets:** raw_orders, daily_revenue (daily partitions from 2025-01-01, group orders)
**Automation:** partitioned schedule 03:00 UTC; late_files sensor with run_key per file
**Checks:** raw_orders_has_rows (blocking)
**Tests:** materialize test for 2025-09-01 passes; definitions load in 2.1 s
**Pending approval:** backfill 2025-01-01..2025-08-31 (243 partitions, pool limit 4)