Dagster Assets: Partitions, Checks, and Secrets That Do Not Live in Ops

We rematerialized fct_orders because Monday’s late arrivals never landed. The asset had no partitions. Dagster did what we asked: it rebuilt the whole table. Eighteen months, 4.2 billion rows, 70 minutes of warehouse queue, and Slack from finance at minute 12. The run was green. The number was wrong for an hour because every downstream reader saw a swap mid-rebuild.

Airflow would have failed the same way if the DAG was one unpartitioned CREATE TABLE AS. The difference is the object you debug. In Airflow you debug a task. In Dagster you debug a table. If you treat assets like tasks, you paid for lineage and got a nicer scheduler. The comparison that decides the tool is in Airflow vs Dagster vs Prefect. This post is what production assets actually require.

flowchart TD
  late[Late Monday rows] --> part{Asset partitioned?}
  part -->|yes| day[Materialize dt=Monday]
  part -->|no| all[Rebuild the whole table]
  day --> check[Blocking asset check]
  all --> check
  check -->|fail| stop[Do not mark materialized]
  check -->|pass| dbt[dbt tests on gold]

Partitions bound the rewrite. Checks bound what “materialized” means. dbt still runs after the warehouse write.

Assets are not tasks with extra metadata

An Airflow task succeeds when the operator exits zero. The table can be empty, yesterday’s partition, or a view that now points at the wrong schema. I have watched that shape stay green for eleven days. The first pipeline tutorial still teaches the task model — your first Airflow DAG — and it is the right model for “run this script at 06:00.” It is the wrong model for “these five tables must exist for this day.”

A Dagster asset is the table. Downstream assets do not start because a job finished. They start because the upstream asset has a new materialization. That is the whole inversion.

from dagster import (
    AssetExecutionContext,
    DailyPartitionsDefinition,
    Definitions,
    asset,
)

days = DailyPartitionsDefinition(start_date="2025-01-01")


@asset(
    partitions_def=days,
    deps=["stg_orders"],
    required_resource_keys={"wh"},
)
def fct_orders(context: AssetExecutionContext) -> None:
    day = context.partition_key  # '2026-08-29', not "all of history"
    context.resources.wh.execute(
        """
        delete from analytics.fct_orders where order_dt = %(day)s;
        insert into analytics.fct_orders
        select order_id, customer_id, amount_usd, order_dt
        from analytics.stg_orders
        where order_dt = %(day)s
        """,
        {"day": day},
    )
    n = context.resources.wh.fetchone(
        "select count(*) from analytics.fct_orders where order_dt = %(day)s",
        {"day": day},
    )[0]
    if n == 0:
        raise RuntimeError(f"fct_orders empty for {day}")

The delete/insert is the same idempotent day rewrite you want in any orchestrator. The partition key is what stops a late Monday from becoming a full-table rebuild. Backfill 90 days and Dagster launches 90 materializations of fct_orders, not one CTAS.

Asset checks are not dbt tests

dbt tests run after the model exists in the warehouse. They are good. They are late. An asset check runs in the materialization and can refuse the “success” the rest of the graph will see.

from dagster import AssetCheckResult, asset_check


@asset_check(asset=fct_orders, blocking=True, required_resource_keys={"wh"})
def fct_orders_volume(context: AssetExecutionContext) -> AssetCheckResult:
    day = context.partition_key
    n = context.resources.wh.fetchone(
        "select count(*) from analytics.fct_orders where order_dt = %(day)s",
        {"day": day},
    )[0]
    # Weekday band. A holiday Tuesday is not an incident.
    ok = 1_000 <= n <= 5_000_000
    return AssetCheckResult(
        passed=ok,
        metadata={"rows": n, "day": day},
        description=None if ok else f"volume {n} outside band",
    )

blocking=True means a failed check is a failed materialization. Downstream gold does not start. That is the contract I want for grain and volume on this partition.

Keep dbt tests. They catch the second writer, the leftover key from last month, and the incremental merge that is unique today and duplicated in the target — the failure mode in dbt incremental models. Asset checks do not make the warehouse omniscient. They make “materialized” mean “this partition is fit to read.”

Resources hold clients. Ops do not hold secrets.

The first Dagster repo I inherited built a Snowflake connector inside the asset with a password from os.environ interpolated into a log line. The event log stored the connection string. The rotation was a Slack message.

from dagster import ConfigurableResource, EnvVar
import snowflake.connector


class SnowflakeWh(ConfigurableResource):
    account: str
    user: str
    password: str
    warehouse: str
    database: str

    def execute(self, sql: str, params: dict | None = None) -> None:
        with snowflake.connector.connect(
            account=self.account,
            user=self.user,
            password=self.password,
            warehouse=self.warehouse,
            database=self.database,
        ) as cx:
            cx.cursor().execute(sql, params or {})

    def fetchone(self, sql: str, params: dict | None = None):
        with snowflake.connector.connect(
            account=self.account,
            user=self.user,
            password=self.password,
            warehouse=self.warehouse,
            database=self.database,
        ) as cx:
            cur = cx.cursor()
            cur.execute(sql, params or {})
            return cur.fetchone()


defs = Definitions(
    assets=[fct_orders],
    asset_checks=[fct_orders_volume],
    resources={
        "wh": SnowflakeWh(
            account="acme-prod",
            user=EnvVar("SNOWFLAKE_USER"),
            password=EnvVar("SNOWFLAKE_PASSWORD"),
            warehouse="TRANSFORM_WH",
            database="ANALYTICS",
        )
    },
)

Tests inject a stub resource. Prod injects EnvVar. The asset body never sees a password. If you need the secret in the op “just this once,” you will paste it into a log within a month.

Failure modes I have paid for

Ops graph, no assets. You get retries and a UI. You do not get lineage or a backfill that means a table. Budget the rewrite.

Unpartitioned facts. A one-day fix rewrites history. Partition on the clock the late data arrives on.

Checks only in dbt. Dagster marks the asset materialized. The warehouse test fails twenty minutes later. Downstream already ran.

Secrets in the asset body. They land in compute logs and in the event log. Resources plus EnvVar are the boundary.

Backfill without a row-count check. Ninety empty partitions can be ninety green materializations.

What to do Monday

Pick the tables that page you. Make each one a partitioned asset with a blocking volume check and a resource-injected client. Leave the 06:00 bash jobs as jobs. Do not migrate Airflow by wrapping operators. The asset is the product. The run is just how it got there.