import tempfile
import dagster as dg
import dataframely as dy
import duckdb
import polars as pl
from dagster_duckdb_polars import DuckDBPolarsIOManager
import dagster_dataframely as dd
from dagster_dataframely.data import ParquetIOManager
from dagster_dataframely.errors import ColumnSchemaErrordd.build_rule_checks reports a schema’s rules as asset checks on an asset that dd.asset did not build. The asset can be a plain @dg.asset, an asset another integration builds, or a dg.AssetSpec for a table written outside Dagster. The checks read the table and write nothing. A row that fails a rule stays in the table, and its check fails.
Pass the schema and the asset key to dd.build_rule_checks:
class EventsSchema(dy.Schema):
event_id = dy.Int64(primary_key=True)
amount = dy.Int64(nullable=False, min=0)
score = dy.Int64(
nullable=True,
max=10,
metadata={"dagster_dataframely": {"severity": {"max": "warn"}}},
)
events = dg.AssetSpec(
"events", metadata=dd.wiring.get_column_schema_metadata(EventsSchema)
).with_io_manager_key("warehouse")
event_checks = dd.build_rule_checks(EventsSchema, asset="events")
{
"step": event_checks.node_def.name,
"checks": [(spec.name, spec.blocking) for spec in event_checks.check_specs],
}{'step': 'events__checks',
'checks': [('dy_schema__columns', True),
('dy_rule__primary_key', True),
('dy_rule__event_id__nullability', True),
('dy_rule__amount__nullability', True),
('dy_rule__amount__min', True),
('dy_rule__score__max', False)]}dd.build_rule_checks returns one dg.AssetChecksDefinition, which goes in dg.Definitions(asset_checks=[...]). Dagster runs every check in it in one step, named <key>__checks unless you pass name=.
The checks are the ones an asset built with dd.asset has: the column-schema check, then the checks for the schema’s rules. check_granularity and max_failure_examples mean what they mean on dd.asset, and Settings lists their sources. dd.build_rule_checks resolves each setting once, when you call it.
The Columns tab reads the asset’s own definition metadata, so dd.build_rule_checks cannot fill it. Pass dd.wiring.get_column_schema_metadata(EventsSchema) as the asset’s metadata, as events does above.
The checks read the table as a Dagster input, through the asset’s own IO manager, as a pl.DataFrame.
So the asset needs an IO manager that loads its table. events sets one with dg.AssetSpec.with_io_manager_key. The checks load an asset with no IO manager key through io_manager. If that IO manager does not store the table, the step fails while loading the input. For an asset another integration builds, such as a dbt model or a dlt resource, set the IO manager key where that integration takes one. The tests of dd.build_rule_checks set the key only with dg.AssetSpec.with_io_manager_key. Open an issue for an integration whose assets the checks cannot load.
The run below checks a DuckDB table that SQL wrote, as a dbt model would. It selects only the checks, as the “Execute all” button on the Checks tab does:
def warehouse_with(columns: str) -> str:
"""Create the `analytics.events` table in a new database, and return the database path."""
database = f"{tempfile.mkdtemp()}/warehouse.duckdb"
with duckdb.connect(database) as connection:
connection.sql("create schema analytics")
connection.sql(f"create table analytics.events ({columns})")
connection.sql("insert into analytics.events values (1, 5, 1), (2, -1, 20)")
return database
def check_run(database: str) -> dg.ExecuteInProcessResult:
return dg.materialize(
[events, event_checks],
resources={
"warehouse": DuckDBPolarsIOManager(database=database, schema="analytics")
},
selection=dg.AssetSelection.checks(event_checks),
raise_on_error=False,
)
result = check_run(warehouse_with("event_id BIGINT, amount BIGINT, score BIGINT"))
{
evaluation.check_name: evaluation.severity.value
for evaluation in result.get_asset_check_evaluations()
if not evaluation.passed
}2026-10-22 02:12:01 +0000 - dagster - DEBUG - __ephemeral_asset_job__ - 254373cc-f4ca-4073-bfb4-8dd8aa5e7487 - 4212 - RUN_START - Started execution of run for "__ephemeral_asset_job__".
2026-10-22 02:12:01 +0000 - dagster - DEBUG - __ephemeral_asset_job__ - 254373cc-f4ca-4073-bfb4-8dd8aa5e7487 - 4212 - ENGINE_EVENT - Executing steps in process (pid: 4212)
2026-10-22 02:12:01 +0000 - dagster - DEBUG - __ephemeral_asset_job__ - 254373cc-f4ca-4073-bfb4-8dd8aa5e7487 - 4212 - events__checks - RESOURCE_INIT_STARTED - Starting initialization of resources [io_manager, warehouse].
2026-10-22 02:12:01 +0000 - dagster - DEBUG - __ephemeral_asset_job__ - 254373cc-f4ca-4073-bfb4-8dd8aa5e7487 - 4212 - events__checks - RESOURCE_INIT_SUCCESS - Finished initialization of resources [io_manager, warehouse].
2026-10-22 02:12:01 +0000 - dagster - DEBUG - __ephemeral_asset_job__ - 254373cc-f4ca-4073-bfb4-8dd8aa5e7487 - 4212 - LOGS_CAPTURED - Capturing logs for process (pid: 4212).
2026-10-22 02:12:01 +0000 - dagster - DEBUG - __ephemeral_asset_job__ - 254373cc-f4ca-4073-bfb4-8dd8aa5e7487 - 4212 - events__checks - STEP_START - Started execution of step "events__checks".
2026-10-22 02:12:01 +0000 - dagster - DEBUG - __ephemeral_asset_job__ - 254373cc-f4ca-4073-bfb4-8dd8aa5e7487 - 4212 - events__checks - LOADED_INPUT - Loaded input "table" using input manager "warehouse"
2026-10-22 02:12:01 +0000 - dagster - DEBUG - __ephemeral_asset_job__ - 254373cc-f4ca-4073-bfb4-8dd8aa5e7487 - 4212 - events__checks - STEP_INPUT - Got input "table" of type "DataFrame". (Type check passed).
2026-10-22 02:12:01 +0000 - dagster - DEBUG - __ephemeral_asset_job__ - 254373cc-f4ca-4073-bfb4-8dd8aa5e7487 - 4212 - events__checks - STEP_OUTPUT - Yielded output "events_dy_schema__columns" of type "Any". (Type check passed).
2026-10-22 02:12:01 +0000 - dagster - DEBUG - __ephemeral_asset_job__ - 254373cc-f4ca-4073-bfb4-8dd8aa5e7487 - 4212 - events__checks - ASSET_CHECK_EVALUATION - Asset check 'dy_schema__columns' on 'events' passed. Description: 'Every column matches.'
2026-10-22 02:12:01 +0000 - dagster - DEBUG - __ephemeral_asset_job__ - 254373cc-f4ca-4073-bfb4-8dd8aa5e7487 - 4212 - events__checks - STEP_OUTPUT - Yielded output "events_dy_rule__primary_key" of type "Any". (Type check passed).
2026-10-22 02:12:01 +0000 - dagster - DEBUG - __ephemeral_asset_job__ - 254373cc-f4ca-4073-bfb4-8dd8aa5e7487 - 4212 - events__checks - ASSET_CHECK_EVALUATION - Asset check 'dy_rule__primary_key' on 'events' passed. Description: 'No row failed.'
2026-10-22 02:12:01 +0000 - dagster - DEBUG - __ephemeral_asset_job__ - 254373cc-f4ca-4073-bfb4-8dd8aa5e7487 - 4212 - events__checks - STEP_OUTPUT - Yielded output "events_dy_rule__event_id__nullability" of type "Any". (Type check passed).
2026-10-22 02:12:01 +0000 - dagster - DEBUG - __ephemeral_asset_job__ - 254373cc-f4ca-4073-bfb4-8dd8aa5e7487 - 4212 - events__checks - ASSET_CHECK_EVALUATION - Asset check 'dy_rule__event_id__nullability' on 'events' passed. Description: 'No row failed.'
2026-10-22 02:12:01 +0000 - dagster - DEBUG - __ephemeral_asset_job__ - 254373cc-f4ca-4073-bfb4-8dd8aa5e7487 - 4212 - events__checks - STEP_OUTPUT - Yielded output "events_dy_rule__amount__nullability" of type "Any". (Type check passed).
2026-10-22 02:12:01 +0000 - dagster - DEBUG - __ephemeral_asset_job__ - 254373cc-f4ca-4073-bfb4-8dd8aa5e7487 - 4212 - events__checks - ASSET_CHECK_EVALUATION - Asset check 'dy_rule__amount__nullability' on 'events' passed. Description: 'No row failed.'
2026-10-22 02:12:01 +0000 - dagster - DEBUG - __ephemeral_asset_job__ - 254373cc-f4ca-4073-bfb4-8dd8aa5e7487 - 4212 - events__checks - STEP_OUTPUT - Yielded output "events_dy_rule__amount__min" of type "Any". (Type check passed).
2026-10-22 02:12:01 +0000 - dagster - DEBUG - __ephemeral_asset_job__ - 254373cc-f4ca-4073-bfb4-8dd8aa5e7487 - 4212 - events__checks - ASSET_CHECK_EVALUATION - Asset check 'dy_rule__amount__min' on 'events' did not pass. Description: '1 row failed.'
2026-10-22 02:12:01 +0000 - dagster - DEBUG - __ephemeral_asset_job__ - 254373cc-f4ca-4073-bfb4-8dd8aa5e7487 - 4212 - events__checks - STEP_OUTPUT - Yielded output "events_dy_rule__score__max" of type "Any". (Type check passed).
2026-10-22 02:12:01 +0000 - dagster - DEBUG - __ephemeral_asset_job__ - 254373cc-f4ca-4073-bfb4-8dd8aa5e7487 - 4212 - events__checks - ASSET_CHECK_EVALUATION - Asset check 'dy_rule__score__max' on 'events' did not pass. Description: '1 row failed.'
2026-10-22 02:12:01 +0000 - dagster - ERROR - __ephemeral_asset_job__ - 254373cc-f4ca-4073-bfb4-8dd8aa5e7487 - 4212 - events__checks - STEP_FAILURE - Execution of step "events__checks" failed.
dagster._core.errors.DagsterAssetCheckFailedError: 1 blocking asset check failed with ERROR severity:
events: dy_rule__amount__min
Stack Trace:
File "/home/runner/work/dagster-dataframely/dagster-dataframely/.venv/lib/python3.12/site-packages/dagster/_core/execution/plan/execute_plan.py", line 243, in dagster_event_sequence_for_step
yield from check.generator(step_events)
File "/home/runner/work/dagster-dataframely/dagster-dataframely/.venv/lib/python3.12/site-packages/dagster/_core/execution/plan/execute_step.py", line 543, in core_dagster_event_sequence_for_step
raise DagsterAssetCheckFailedError(
/home/runner/work/dagster-dataframely/dagster-dataframely/.venv/lib/python3.12/site-packages/dagster/_core/execution/context_creation_job.py:273: RuntimeWarning: coroutine 'BaseEventLoop.shutdown_asyncgens' was never awaited
pass
RuntimeWarning: Enable tracemalloc to get the object allocation traceback
2026-10-22 02:12:01 +0000 - dagster - DEBUG - __ephemeral_asset_job__ - 254373cc-f4ca-4073-bfb4-8dd8aa5e7487 - 4212 - ENGINE_EVENT - Finished steps in process (pid: 4212) in 96ms
2026-10-22 02:12:01 +0000 - dagster - ERROR - __ephemeral_asset_job__ - 254373cc-f4ca-4073-bfb4-8dd8aa5e7487 - 4212 - RUN_FAILURE - Execution of run for "__ephemeral_asset_job__" failed. Steps failed: ['events__checks'].{'dy_rule__amount__min': 'ERROR', 'dy_rule__score__max': 'WARN'}Row 2 fails amount|min, an "error" rule, and score|max, a "warn" rule.
A failing "error" rule’s check stops the assets downstream. Each "error" rule’s check is blocking, as it is for dd.asset with filter=False. So Dagster fails the check step and does not run the assets downstream of events:
result.successFalsedd.build_rule_checks takes no filter argument, and it writes no error rows.
A failing "warn" rule’s check only reports. Declare a rule "warn" to report its failures without stopping anything. Rule severity lists the ways to declare it.
The check results have the same descriptions and metadata as the results of an asset built with dd.asset. What a run produces lists them.
A run writes the failure examples to the Dagster event log unredacted.
A table that SQL wrote can store a different dtype from the one the schema declares. DuckDB’s INTEGER reads back as Int32, so it does not match dy.Int64:
try:
dg.materialize(
[events, event_checks],
resources={
"warehouse": DuckDBPolarsIOManager(
database=warehouse_with(
"event_id BIGINT, amount INTEGER, score BIGINT"
),
schema="analytics",
)
},
selection=dg.AssetSelection.checks(event_checks),
)
except ColumnSchemaError as error:
print(error)2026-10-22 02:12:01 +0000 - dagster - DEBUG - __ephemeral_asset_job__ - 8a47920e-1c00-4bad-a9f8-abe27ad7f3dc - 4212 - RUN_START - Started execution of run for "__ephemeral_asset_job__".
2026-10-22 02:12:01 +0000 - dagster - DEBUG - __ephemeral_asset_job__ - 8a47920e-1c00-4bad-a9f8-abe27ad7f3dc - 4212 - ENGINE_EVENT - Executing steps in process (pid: 4212)
2026-10-22 02:12:01 +0000 - dagster - DEBUG - __ephemeral_asset_job__ - 8a47920e-1c00-4bad-a9f8-abe27ad7f3dc - 4212 - events__checks - RESOURCE_INIT_STARTED - Starting initialization of resources [io_manager, warehouse].
2026-10-22 02:12:01 +0000 - dagster - DEBUG - __ephemeral_asset_job__ - 8a47920e-1c00-4bad-a9f8-abe27ad7f3dc - 4212 - events__checks - RESOURCE_INIT_SUCCESS - Finished initialization of resources [io_manager, warehouse].
2026-10-22 02:12:01 +0000 - dagster - DEBUG - __ephemeral_asset_job__ - 8a47920e-1c00-4bad-a9f8-abe27ad7f3dc - 4212 - LOGS_CAPTURED - Capturing logs for process (pid: 4212).
2026-10-22 02:12:01 +0000 - dagster - DEBUG - __ephemeral_asset_job__ - 8a47920e-1c00-4bad-a9f8-abe27ad7f3dc - 4212 - events__checks - STEP_START - Started execution of step "events__checks".
2026-10-22 02:12:01 +0000 - dagster - DEBUG - __ephemeral_asset_job__ - 8a47920e-1c00-4bad-a9f8-abe27ad7f3dc - 4212 - events__checks - LOADED_INPUT - Loaded input "table" using input manager "warehouse"
2026-10-22 02:12:01 +0000 - dagster - DEBUG - __ephemeral_asset_job__ - 8a47920e-1c00-4bad-a9f8-abe27ad7f3dc - 4212 - events__checks - STEP_INPUT - Got input "table" of type "DataFrame". (Type check passed).
2026-10-22 02:12:01 +0000 - dagster - DEBUG - __ephemeral_asset_job__ - 8a47920e-1c00-4bad-a9f8-abe27ad7f3dc - 4212 - events__checks - STEP_OUTPUT - Yielded output "events_dy_schema__columns" of type "Any". (Type check passed).
2026-10-22 02:12:01 +0000 - dagster - DEBUG - __ephemeral_asset_job__ - 8a47920e-1c00-4bad-a9f8-abe27ad7f3dc - 4212 - events__checks - ASSET_CHECK_EVALUATION - Asset check 'dy_schema__columns' on 'events' did not pass. Description: '1 column does not match: amount.'
2026-10-22 02:12:01 +0000 - dagster - ERROR - __ephemeral_asset_job__ - 8a47920e-1c00-4bad-a9f8-abe27ad7f3dc - 4212 - events__checks - STEP_FAILURE - Execution of step "events__checks" failed.
dagster_dataframely.errors.ColumnSchemaError: Column 'amount' (expected Int64, got Int32) does not match EventsSchema. Declare the dtype the table stores in EventsSchema, or change what writes the table. This package never casts for you.
Stack Trace:
File "/home/runner/work/dagster-dataframely/dagster-dataframely/.venv/lib/python3.12/site-packages/dagster/_core/execution/plan/execute_plan.py", line 243, in dagster_event_sequence_for_step
yield from check.generator(step_events)
File "/home/runner/work/dagster-dataframely/dagster-dataframely/.venv/lib/python3.12/site-packages/dagster/_core/execution/plan/execute_step.py", line 505, in core_dagster_event_sequence_for_step
for user_event in _step_output_error_checked_user_event_sequence(
File "/home/runner/work/dagster-dataframely/dagster-dataframely/.venv/lib/python3.12/site-packages/dagster/_core/execution/plan/execute_step.py", line 184, in _step_output_error_checked_user_event_sequence
for user_event in user_event_sequence:
File "/home/runner/work/dagster-dataframely/dagster-dataframely/.venv/lib/python3.12/site-packages/dagster/_core/execution/plan/execute_step.py", line 88, in _process_asset_results_to_events
for user_event in user_event_sequence:
File "/home/runner/work/dagster-dataframely/dagster-dataframely/.venv/lib/python3.12/site-packages/dagster/_core/execution/plan/compute.py", line 188, in execute_core_compute
for step_output in _yield_compute_results(step_context, inputs, compute_fn, compute_context):
File "/home/runner/work/dagster-dataframely/dagster-dataframely/.venv/lib/python3.12/site-packages/dagster/_core/execution/plan/compute.py", line 157, in _yield_compute_results
for event in iterate_with_context(
File "/home/runner/work/dagster-dataframely/dagster-dataframely/.venv/lib/python3.12/site-packages/dagster/_utils/__init__.py", line 394, in iterate_with_context
next_output = next(iterator)
^^^^^^^^^^^^^^
File "/home/runner/work/dagster-dataframely/dagster-dataframely/src/dagster_dataframely/_rule_checks.py", line 106, in _results
raise ColumnSchemaError(
/home/runner/work/dagster-dataframely/dagster-dataframely/.venv/lib/python3.12/site-packages/dagster/_core/execution/context_creation_job.py:273: RuntimeWarning: coroutine 'BaseEventLoop.shutdown_asyncgens' was never awaited
pass
RuntimeWarning: Enable tracemalloc to get the object allocation traceback
2026-10-22 02:12:01 +0000 - dagster - ERROR - __ephemeral_asset_job__ - 8a47920e-1c00-4bad-a9f8-abe27ad7f3dc - 4212 - RUN_FAILURE - Execution of run for "__ephemeral_asset_job__" failed. An exception was thrown during execution.
dagster_dataframely.errors.ColumnSchemaError: Column 'amount' (expected Int64, got Int32) does not match EventsSchema. Declare the dtype the table stores in EventsSchema, or change what writes the table. This package never casts for you.
Stack Trace:
File "/home/runner/work/dagster-dataframely/dagster-dataframely/.venv/lib/python3.12/site-packages/dagster/_core/execution/api.py", line 742, in job_execution_iterator
for event in job_context.executor.execute(job_context, execution_plan):
File "/home/runner/work/dagster-dataframely/dagster-dataframely/.venv/lib/python3.12/site-packages/dagster/_core/executor/in_process.py", line 71, in execute
yield from iter(
File "/home/runner/work/dagster-dataframely/dagster-dataframely/.venv/lib/python3.12/site-packages/dagster/_core/execution/api.py", line 873, in __iter__
yield from self.iterator(
File "/home/runner/work/dagster-dataframely/dagster-dataframely/.venv/lib/python3.12/site-packages/dagster/_core/executor/in_process.py", line 28, in inprocess_execution_iterator
yield from inner_plan_execution_iterator(
File "/home/runner/work/dagster-dataframely/dagster-dataframely/.venv/lib/python3.12/site-packages/dagster/_core/execution/plan/execute_plan.py", line 85, in inner_plan_execution_iterator
for step_event in check.generator(dagster_event_sequence_for_step(step_context)):
File "/home/runner/work/dagster-dataframely/dagster-dataframely/.venv/lib/python3.12/site-packages/dagster/_core/execution/plan/execute_plan.py", line 303, in dagster_event_sequence_for_step
raise failure
File "/home/runner/work/dagster-dataframely/dagster-dataframely/.venv/lib/python3.12/site-packages/dagster/_core/execution/plan/execute_plan.py", line 243, in dagster_event_sequence_for_step
yield from check.generator(step_events)
File "/home/runner/work/dagster-dataframely/dagster-dataframely/.venv/lib/python3.12/site-packages/dagster/_core/execution/plan/execute_step.py", line 505, in core_dagster_event_sequence_for_step
for user_event in _step_output_error_checked_user_event_sequence(
File "/home/runner/work/dagster-dataframely/dagster-dataframely/.venv/lib/python3.12/site-packages/dagster/_core/execution/plan/execute_step.py", line 184, in _step_output_error_checked_user_event_sequence
for user_event in user_event_sequence:
File "/home/runner/work/dagster-dataframely/dagster-dataframely/.venv/lib/python3.12/site-packages/dagster/_core/execution/plan/execute_step.py", line 88, in _process_asset_results_to_events
for user_event in user_event_sequence:
File "/home/runner/work/dagster-dataframely/dagster-dataframely/.venv/lib/python3.12/site-packages/dagster/_core/execution/plan/compute.py", line 188, in execute_core_compute
for step_output in _yield_compute_results(step_context, inputs, compute_fn, compute_context):
File "/home/runner/work/dagster-dataframely/dagster-dataframely/.venv/lib/python3.12/site-packages/dagster/_core/execution/plan/compute.py", line 157, in _yield_compute_results
for event in iterate_with_context(
File "/home/runner/work/dagster-dataframely/dagster-dataframely/.venv/lib/python3.12/site-packages/dagster/_utils/__init__.py", line 394, in iterate_with_context
next_output = next(iterator)
^^^^^^^^^^^^^^
File "/home/runner/work/dagster-dataframely/dagster-dataframely/src/dagster_dataframely/_rule_checks.py", line 106, in _results
raise ColumnSchemaError(
Column 'amount' (expected Int64, got Int32) does not match EventsSchema. Declare the dtype the table stores in EventsSchema, or change what writes the table. This package never casts for you.The column-schema check fails, and the step fails with ColumnSchemaError. The checks evaluate no rule. dd.build_rule_checks has no decorated function to cast in.
dd.build_rule_checks also checks a plain @dg.asset. Use it when the write must not depend on the check results:
@dg.asset(metadata=dd.wiring.get_column_schema_metadata(EventsSchema))
def staged_events() -> pl.DataFrame:
return pl.DataFrame({"event_id": [1, 2], "amount": [5, -1], "score": [1, 2]})
staged_checks = dd.build_rule_checks(EventsSchema, asset="staged_events")
warehouse_orders shows checks that run in their own step, after the IO manager writes the table. The selected check is blocking because its rule is "error", and the rows that failed it are in the table, not in the error rows.The IO manager writes whatever the asset returns, and the checks run afterwards against the written table. Unlike with dd.asset, the table can contain rows that failed an "error" rule. Only a failing check reports them.
Pass the asset’s partitions_def, and each run checks one partition:
days = dg.StaticPartitionsDefinition(["2026-10-01", "2026-10-02"])
@dg.asset(
partitions_def=days, metadata=dd.wiring.get_column_schema_metadata(EventsSchema)
)
def daily_events(context: dg.AssetExecutionContext) -> pl.DataFrame:
amount = 5 if context.partition_key == "2026-10-01" else -1
return pl.DataFrame({"event_id": [1], "amount": [amount], "score": [1]})
daily_checks = dd.build_rule_checks(
EventsSchema, asset="daily_events", partitions_def=days
)/home/runner/work/dagster-dataframely/dagster-dataframely/src/dagster_dataframely/_checks.py:137: PreviewWarning: Specifying a partitions_def on an AssetCheckSpec is currently in preview, and may have breaking changes in patch version releases. This feature is not considered ready for production use.
column_schema = dg.AssetCheckSpec(
/home/runner/work/dagster-dataframely/dagster-dataframely/src/dagster_dataframely/_checks.py:147: PreviewWarning: Specifying a partitions_def on an AssetCheckSpec is currently in preview, and may have breaking changes in patch version releases. This feature is not considered ready for production use.
dg.AssetCheckSpec(The checks read only that partition’s rows, and Dagster records each evaluation under the partition. So the checks validate each partition separately, as an asset built with dd.asset does. Nothing checks that partitions_def matches the asset’s. Without partitions_def, the checks read every partition at once. With dg.FilesystemIOManager, the input is a dict, and Dagster fails the step with DagsterTypeCheckDidNotPass.
A run can select any of the checks, and it reports only those. Each run still validates the whole table, or the whole partition, so selecting one check takes as long as selecting all of them.