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

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude <noreply@anthropic.com>

Author: prha

python_modules/dagster/dagster/_core/definitions/asset_checks/asset_check_result.py

@@ -172,6 +172,21 @@ def to_asset_check_evaluation(
         else:
             target_materialization_data = None
 
+        if step_context.has_partition_key:
+            assets_def = step_context.job_def.asset_layer.get_assets_def_for_node(
+                step_context.node_handle
+            )
+            assert assets_def
+            spec = assets_def_for_check.get_spec_for_check_key(check_key)
+            if spec.partitions_def is not None and spec.partitions_def.has_partition_key(
+                step_context.partition_key
+            ):
+                partition = step_context.partition_key
+            else:
+                partition = None
+        else:
+            partition = None
+
         return AssetCheckEvaluation(
             check_name=check_key.name,
             asset_key=check_key.asset_key,
@@ -181,6 +196,7 @@ def to_asset_check_evaluation(
             severity=self.severity,
             description=self.description,
             blocking=assets_def_for_check.get_spec_for_check_key(check_key).blocking,
+            partition=partition,
         )
 
     def with_metadata(self, metadata: Mapping[str, RawMetadataValue]) -> "AssetCheckResult":  # pyright: ignore[reportIncompatibleMethodOverride]

python_modules/dagster/dagster/_core/definitions/asset_checks/asset_check_spec.py

@@ -13,6 +13,7 @@
 
 from dagster._annotations import PublicAttr, public
 from dagster._core.definitions.asset_key import AssetCheckKey, AssetKey, CoercibleToAssetKey
+from dagster._core.definitions.partitions.definition import PartitionsDefinition
 
 if TYPE_CHECKING:
     from dagster._core.definitions.assets.definition.asset_dep import AssetDep, CoercibleToAssetDep
@@ -58,6 +59,7 @@ class AssetCheckSpec(IHaveNew, LegacyNamedTupleMixin):
     blocking: PublicAttr[bool]
     metadata: PublicAttr[Mapping[str, Any]]
     automation_condition: PublicAttr[Optional[LazyAutomationCondition]]
+    partitions_def: PublicAttr[Optional[PartitionsDefinition]]
 
     """Defines information about an asset check, except how to execute it.
 
@@ -80,6 +82,7 @@ class AssetCheckSpec(IHaveNew, LegacyNamedTupleMixin):
             that multi-asset is responsible for enforcing that downstream assets within the
             same step do not execute after a blocking asset check fails.
         metadata (Optional[Mapping[str, Any]]):  A dict of static metadata for this asset check.
+        partitions_def (Optional[PartitionsDefinition]): The partitions definition for this asset check.
     """
 
     def __new__(
@@ -92,6 +95,7 @@ def __new__(
         blocking: bool = False,
         metadata: Optional[Mapping[str, Any]] = None,
         automation_condition: Optional["AutomationCondition[AssetCheckKey]"] = None,
+        partitions_def: Optional[PartitionsDefinition] = None,
     ):
         from dagster._core.definitions.assets.definition.asset_dep import (
             coerce_to_deps_and_check_duplicates,
@@ -119,6 +123,7 @@ def __new__(
             blocking=blocking,
             metadata=metadata or {},
             automation_condition=automation_condition,
+            partitions_def=partitions_def,
         )
 
     def get_python_identifier(self) -> str:

python_modules/dagster/dagster/_core/definitions/decorators/asset_check_decorator.py

@@ -28,6 +28,7 @@
 from dagster._core.definitions.decorators.op_decorator import _Op
 from dagster._core.definitions.events import AssetKey, CoercibleToAssetKey
 from dagster._core.definitions.output import Out
+from dagster._core.definitions.partitions.definition import PartitionsDefinition
 from dagster._core.definitions.policy import RetryPolicy
 from dagster._core.definitions.source_asset import SourceAsset
 from dagster._core.definitions.utils import DEFAULT_OUTPUT
@@ -113,6 +114,7 @@ def asset_check(
     metadata: Optional[Mapping[str, Any]] = None,
     automation_condition: Optional[AutomationCondition[AssetCheckKey]] = None,
     pool: Optional[str] = None,
+    partitions_def: Optional[PartitionsDefinition] = None,
 ) -> Callable[[AssetCheckFunction], AssetChecksDefinition]:
     """Create a definition for how to execute an asset check.
 
@@ -218,6 +220,7 @@ def inner(fn: AssetCheckFunction) -> AssetChecksDefinition:
             blocking=blocking,
             metadata=metadata,
             automation_condition=automation_condition,
+            partitions_def=partitions_def,
         )
 
         resource_defs_for_execution = wrap_resources_for_execution(resource_defs)

python_modules/dagster/dagster/_core/events/__init__.py

@@ -39,6 +39,7 @@
     AssetCheckEvaluation,
     AssetCheckEvaluationPlanned,
 )
+from dagster._core.definitions.asset_checks.asset_check_spec import AssetCheckKey
 from dagster._core.definitions.asset_health.asset_health import AssetHealthStatus
 from dagster._core.definitions.events import (
     AssetLineageInfo,

python_modules/dagster/dagster/_core/execution/context/asset_check_execution_context.py

@@ -6,6 +6,8 @@
 from dagster._core.definitions.asset_checks.asset_check_spec import AssetCheckKey, AssetCheckSpec
 from dagster._core.definitions.job_definition import JobDefinition
 from dagster._core.definitions.op_definition import OpDefinition
+from dagster._core.definitions.partitions.partition_key_range import PartitionKeyRange
+from dagster._core.definitions.partitions.utils import TimeWindow
 from dagster._core.definitions.repository_definition.repository_definition import (
     RepositoryDefinition,
 )
@@ -135,6 +137,43 @@ def step_launcher(self) -> Optional[StepLauncher]:
     def get_step_execution_context(self) -> StepExecutionContext:
         return self.op_execution_context.get_step_execution_context()
 
+    #### partition related
+    @public
+    @property
+    @_copy_docs_from_op_execution_context
+    def has_partition_key(self) -> bool:
+        return self.op_execution_context.has_partition_key
+
+    @public
+    @property
+    @_copy_docs_from_op_execution_context
+    def partition_key(self) -> str:
+        return self.op_execution_context.partition_key
+
+    @public
+    @property
+    @_copy_docs_from_op_execution_context
+    def partition_keys(self) -> Sequence[str]:
+        return self.op_execution_context.partition_keys
+
+    @public
+    @property
+    @_copy_docs_from_op_execution_context
+    def has_partition_key_range(self) -> bool:
+        return self.op_execution_context.has_partition_key_range
+
+    @public
+    @property
+    @_copy_docs_from_op_execution_context
+    def partition_key_range(self) -> PartitionKeyRange:
+        return self.op_execution_context.partition_key_range
+
+    @public
+    @property
+    @_copy_docs_from_op_execution_context
+    def partition_time_window(self) -> TimeWindow:
+        return self.op_execution_context.partition_time_window
+
     # misc
 
     @public