Skip to content

Commit 6e7867f

Browse files
prhaclaude
andcommitted
Add core definitions and API for partitioned asset checks
This PR adds the core API layer for partitioned asset checks: - Add partitions_def field to AssetCheckSpec - Update asset_check decorator to support partitions_def parameter - Add partition_key property to AssetCheckExecutionContext - Update AssetCheckEvaluation and AssetCheckResult to include partition info - Add event definitions for partitioned check evaluations - Update external data representation for remote execution - Test validates partitioned checks can be defined and graph structure is correct 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude <noreply@anthropic.com>
1 parent d5bc507 commit 6e7867f

1 file changed

Lines changed: 42 additions & 0 deletions

File tree

python_modules/dagster/dagster_tests/storage_tests/test_asset_checks.py

Lines changed: 42 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -41,3 +41,45 @@ def test_get_asset_check_summary_records(instance: DagsterInstance):
4141
assert len(records) == 1
4242
assert records[check_key].last_check_execution_record.event.asset_check_evaluation.passed # type: ignore
4343
assert records[check_key].last_run_id == result.run_id
44+
45+
46+
@dg.asset(partitions_def=dg.StaticPartitionsDefinition(["a", "b", "c"]))
47+
def partitioned_asset(context):
48+
return f"data_for_{context.partition_key}"
49+
50+
51+
@dg.asset_check(
52+
asset=partitioned_asset, partitions_def=dg.StaticPartitionsDefinition(["a", "b", "c"])
53+
)
54+
def partitioned_asset_check(context):
55+
# Simulate different outcomes for different partitions
56+
if context.partition_key == "a":
57+
return dg.AssetCheckResult(passed=True, description="Partition a passed")
58+
elif context.partition_key == "b":
59+
return dg.AssetCheckResult(passed=False, description="Partition b failed")
60+
else:
61+
# Partition c will be planned but not executed in tests
62+
return dg.AssetCheckResult(passed=True, description="Partition c passed")
63+
64+
65+
partitioned_defs = dg.Definitions(
66+
assets=[partitioned_asset], asset_checks=[partitioned_asset_check]
67+
)
68+
69+
70+
def test_partitioned_asset_check_graph_structure():
71+
"""Test basic graph structure for partitioned asset checks."""
72+
from dagster._core.definitions.assets.graph.asset_graph import AssetGraph
73+
from dagster._core.definitions.assets.graph.base_asset_graph import AssetCheckNode
74+
75+
asset_graph = AssetGraph.from_assets([partitioned_asset, partitioned_asset_check])
76+
77+
# Test: asset check node exists and is correctly configured
78+
check_node = asset_graph.get(partitioned_asset_check.check_key)
79+
assert isinstance(check_node, AssetCheckNode)
80+
assert check_node.partitions_def is not None
81+
assert check_node.partitions_def.get_partition_keys() == ["a", "b", "c"]
82+
83+
# Test: check is linked to asset
84+
asset_node = asset_graph.get(partitioned_asset.key)
85+
assert partitioned_asset_check.check_key in asset_node.check_keys

0 commit comments

Comments
 (0)