dagster-io
diff --git a/‎python_modules/libraries/dagster-dg-cli/dagster_dg_cli/api_layer/api/asset.py‎
Lines changed: 7 additions & 0 deletions b/‎python_modules/libraries/dagster-dg-cli/dagster_dg_cli/api_layer/api/asset.py‎
Lines changed: 7 additions & 0 deletions
diff --git a/‎python_modules/libraries/dagster-dg-cli/dagster_dg_cli/api_layer/api/asset_check.py‎
Lines changed: 44 additions & 0 deletions b/‎python_modules/libraries/dagster-dg-cli/dagster_dg_cli/api_layer/api/asset_check.py‎
Lines changed: 44 additions & 0 deletions
diff --git a/‎python_modules/libraries/dagster-dg-cli/dagster_dg_cli/api_layer/graphql_adapter/asset.py‎
Lines changed: 53 additions & 0 deletions b/‎python_modules/libraries/dagster-dg-cli/dagster_dg_cli/api_layer/graphql_adapter/asset.py‎
Lines changed: 53 additions & 0 deletions
diff --git a/‎python_modules/libraries/dagster-dg-cli/dagster_dg_cli/api_layer/graphql_adapter/asset_check.py‎
Lines changed: 166 additions & 0 deletions b/‎python_modules/libraries/dagster-dg-cli/dagster_dg_cli/api_layer/graphql_adapter/asset_check.py‎
Lines changed: 166 additions & 0 deletions
diff --git a/‎python_modules/libraries/dagster-dg-cli/dagster_dg_cli/api_layer/schemas/asset.py‎
Lines changed: 9 additions & 0 deletions b/‎python_modules/libraries/dagster-dg-cli/dagster_dg_cli/api_layer/schemas/asset.py‎
Lines changed: 9 additions & 0 deletions
diff --git a/‎python_modules/libraries/dagster-dg-cli/dagster_dg_cli/api_layer/schemas/asset_check.py‎
Lines changed: 53 additions & 0 deletions b/‎python_modules/libraries/dagster-dg-cli/dagster_dg_cli/api_layer/schemas/asset_check.py‎
Lines changed: 53 additions & 0 deletions
@@ -8,6 +8,7 @@
     get_asset_evaluations_via_graphql,
     get_asset_events_via_graphql,
     get_asset_health_via_graphql,
+    get_asset_partition_status_via_graphql,
     get_dg_plus_api_asset_via_graphql,
     list_dg_plus_api_assets_via_graphql,
 )
@@ -20,6 +21,7 @@
         DgApiAssetList,
         DgApiAssetStatus,
         DgApiEvaluationRecordList,
+        DgApiPartitionStats,
     )
 
 
@@ -116,3 +118,8 @@ def get_evaluations(
             cursor=cursor,
             include_nodes=include_nodes,
         )
+
+    def get_partition_status(self, asset_key: str) -> "DgApiPartitionStats":
+        """Get partition materialization stats for an asset by slash-separated key."""
+        asset_key_parts = asset_key.split("/")
+        return get_asset_partition_status_via_graphql(self.client, asset_key_parts)
@@ -0,0 +1,44 @@
+"""Asset check API implementation."""
+
+from dataclasses import dataclass
+from typing import TYPE_CHECKING
+
+from dagster_dg_cli.api_layer.graphql_adapter.asset_check import (
+    get_asset_check_executions_via_graphql,
+    list_asset_checks_via_graphql,
+)
+from dagster_dg_cli.utils.plus.gql_client import IGraphQLClient
+
+if TYPE_CHECKING:
+    from dagster_dg_cli.api_layer.schemas.asset_check import (
+        DgApiAssetCheckExecutionList,
+        DgApiAssetCheckList,
+    )
+
+
+@dataclass(frozen=True)
+class DgApiAssetCheckApi:
+    """API for asset check operations."""
+
+    client: IGraphQLClient
+
+    def list_asset_checks(self, asset_key: str) -> "DgApiAssetCheckList":
+        """List asset checks for a given asset key."""
+        return list_asset_checks_via_graphql(self.client, asset_key)
+
+    def get_check_executions(
+        self,
+        *,
+        asset_key: str,
+        check_name: str,
+        limit: int = 25,
+        cursor: str | None = None,
+    ) -> "DgApiAssetCheckExecutionList":
+        """Get execution history for an asset check."""
+        return get_asset_check_executions_via_graphql(
+            self.client,
+            asset_key=asset_key,
+            check_name=check_name,
+            limit=limit,
+            cursor=cursor,
+        )
@@ -19,6 +19,7 @@
     DgApiEvaluationRecordList,
     DgApiPartitionDefinition,
     DgApiPartitionMapping,
+    DgApiPartitionStats,
 )
 from dagster_dg_cli.utils.plus.gql_client import IGraphQLClient
 
@@ -786,3 +787,55 @@ def get_asset_evaluations_via_graphql(
         )
 
     return DgApiEvaluationRecordList(items=evaluations)
+
+
+# ---------------------------------------------------------------------------
+# Partition status
+# ---------------------------------------------------------------------------
+
+ASSET_PARTITION_STATUS_QUERY = """
+query AssetPartitionStatusQuery($assetKey: AssetKeyInput!) {
+    assetNodeOrError(assetKey: $assetKey) {
+        __typename
+        ... on AssetNode {
+            partitionStats {
+                numMaterialized
+                numPartitions
+                numFailed
+                numMaterializing
+            }
+        }
+        ... on AssetNotFoundError {
+            message
+        }
+    }
+}
+"""
+
+
+def get_asset_partition_status_via_graphql(
+    client: IGraphQLClient, asset_key_parts: list[str]
+) -> DgApiPartitionStats:
+    """Fetch partition materialization stats for an asset."""
+    variables = {"assetKey": {"path": asset_key_parts}}
+    result = client.execute(ASSET_PARTITION_STATUS_QUERY, variables)
+
+    asset_node = result.get("assetNodeOrError")
+    if not asset_node:
+        raise Exception("No asset data in GraphQL response")
+
+    typename = asset_node.get("__typename")
+    if typename != "AssetNode":
+        error_msg = asset_node.get("message", f"GraphQL error: {typename}")
+        raise Exception(error_msg)
+
+    partition_stats = asset_node.get("partitionStats")
+    if not partition_stats:
+        raise Exception("Asset does not have partitions")
+
+    return DgApiPartitionStats(
+        num_materialized=partition_stats["numMaterialized"],
+        num_failed=partition_stats["numFailed"],
+        num_materializing=partition_stats["numMaterializing"],
+        num_partitions=partition_stats["numPartitions"],
+    )
@@ -0,0 +1,166 @@
+"""GraphQL implementation for asset check operations."""
+
+from typing import TYPE_CHECKING, Any
+
+from dagster_dg_cli.utils.plus.gql_client import IGraphQLClient
+
+if TYPE_CHECKING:
+    from dagster_dg_cli.api_layer.schemas.asset_check import (
+        DgApiAssetCheckExecutionList,
+        DgApiAssetCheckList,
+    )
+
+ASSET_CHECKS_QUERY = """
+query AssetChecksQuery($assetKey: AssetKeyInput!) {
+    assetChecksOrError(assetKey: $assetKey) {
+        __typename
+        ... on AssetChecks {
+            checks {
+                name
+                description
+                blocking
+                jobNames
+                canExecuteIndividually
+                assetKey {
+                    path
+                }
+            }
+        }
+        ... on NeedsMigration {
+            message
+        }
+        ... on NeedsUserCodeUpgrade {
+            message
+        }
+        ... on NeedsAgentUpgrade {
+            message
+        }
+    }
+}
+"""
+
+ASSET_CHECK_EXECUTIONS_QUERY = """
+query AssetCheckExecutionsQuery(
+    $assetKey: AssetKeyInput!,
+    $checkName: String!,
+    $limit: Int!,
+    $cursor: String
+) {
+    assetCheckExecutions(
+        assetKey: $assetKey,
+        checkName: $checkName,
+        limit: $limit,
+        cursor: $cursor
+    ) {
+        id
+        runId
+        status
+        timestamp
+        stepKey
+        evaluation {
+            severity
+            timestamp
+            targetMaterialization {
+                runId
+                timestamp
+            }
+            metadataEntries {
+                label
+            }
+        }
+    }
+}
+"""
+
+
+def process_asset_checks_response(
+    graphql_response: dict[str, Any], asset_key: str
+) -> "DgApiAssetCheckList":
+    """Process GraphQL response into DgApiAssetCheckList."""
+    from dagster_dg_cli.api_layer.schemas.asset_check import DgApiAssetCheck, DgApiAssetCheckList
+
+    checks_result = graphql_response.get("assetChecksOrError")
+    if not checks_result:
+        raise Exception("No asset checks data in GraphQL response")
+
+    typename = checks_result.get("__typename")
+    if typename != "AssetChecks":
+        error_msg = checks_result.get("message", f"GraphQL error: {typename}")
+        raise Exception(error_msg)
+
+    checks_data = checks_result.get("checks", [])
+
+    checks = []
+    for c in checks_data:
+        check_asset_key = "/".join(c.get("assetKey", {}).get("path", []))
+        checks.append(
+            DgApiAssetCheck(
+                name=c["name"],
+                asset_key=check_asset_key or asset_key,
+                description=c.get("description"),
+                blocking=c.get("blocking", False),
+                job_names=c.get("jobNames", []),
+                can_execute_individually=c.get("canExecuteIndividually"),
+            )
+        )
+
+    return DgApiAssetCheckList(items=checks)
+
+
+def process_asset_check_executions_response(
+    graphql_response: dict[str, Any], asset_key: str, check_name: str
+) -> "DgApiAssetCheckExecutionList":
+    """Process GraphQL response into DgApiAssetCheckExecutionList."""
+    from dagster_dg_cli.api_layer.schemas.asset_check import (
+        DgApiAssetCheckExecution,
+        DgApiAssetCheckExecutionList,
+        DgApiAssetCheckExecutionStatus,
+    )
+
+    executions_data = graphql_response.get("assetCheckExecutions", [])
+
+    executions = []
+    for e in executions_data:
+        executions.append(
+            DgApiAssetCheckExecution(
+                id=e["id"],
+                run_id=e["runId"],
+                status=DgApiAssetCheckExecutionStatus(e["status"]),
+                timestamp=float(e["timestamp"]),
+                step_key=e.get("stepKey"),
+                check_name=check_name,
+                asset_key=asset_key,
+            )
+        )
+
+    return DgApiAssetCheckExecutionList(items=executions)
+
+
+def list_asset_checks_via_graphql(client: IGraphQLClient, asset_key: str) -> "DgApiAssetCheckList":
+    """Fetch asset checks using GraphQL."""
+    asset_key_parts = asset_key.split("/")
+    variables = {"assetKey": {"path": asset_key_parts}}
+    result = client.execute(ASSET_CHECKS_QUERY, variables)
+    return process_asset_checks_response(result, asset_key)
+
+
+def get_asset_check_executions_via_graphql(
+    client: IGraphQLClient,
+    *,
+    asset_key: str,
+    check_name: str,
+    limit: int = 25,
+    cursor: str | None = None,
+) -> "DgApiAssetCheckExecutionList":
+    """Fetch asset check executions using GraphQL."""
+    asset_key_parts = asset_key.split("/")
+    variables: dict[str, Any] = {
+        "assetKey": {"path": asset_key_parts},
+        "checkName": check_name,
+        "limit": limit,
+    }
+    if cursor:
+        variables["cursor"] = cursor
+
+    result = client.execute(ASSET_CHECK_EXECUTIONS_QUERY, variables)
+    return process_asset_check_executions_response(result, asset_key, check_name)
@@ -181,3 +181,12 @@ class DgApiEvaluationRecordList(BaseModel):
     """GET /api/assets/{key}/evaluations response."""
 
     items: list[DgApiEvaluationRecord]
+
+
+class DgApiPartitionStats(BaseModel):
+    """Partition materialization statistics for an asset."""
+
+    num_materialized: int
+    num_failed: int
+    num_materializing: int
+    num_partitions: int
@@ -0,0 +1,53 @@
+"""Asset check schema definitions."""
+
+from enum import Enum
+
+from pydantic import BaseModel
+
+
+class DgApiAssetCheckExecutionStatus(str, Enum):
+    """Asset check execution status."""
+
+    IN_PROGRESS = "IN_PROGRESS"
+    SUCCEEDED = "SUCCEEDED"
+    FAILED = "FAILED"
+    SKIPPED = "SKIPPED"
+    EXECUTION_FAILED = "EXECUTION_FAILED"
+
+
+class DgApiAssetCheck(BaseModel):
+    """Single asset check metadata model."""
+
+    name: str
+    asset_key: str
+    description: str | None = None
+    blocking: bool = False
+    job_names: list[str] = []
+    can_execute_individually: str | None = None
+
+
+class DgApiAssetCheckList(BaseModel):
+    """List of asset checks response."""
+
+    items: list[DgApiAssetCheck]
+
+
+class DgApiAssetCheckExecution(BaseModel):
+    """Single asset check execution record."""
+
+    id: str
+    run_id: str
+    status: DgApiAssetCheckExecutionStatus
+    timestamp: float
+    partition: str | None = None
+    step_key: str | None = None
+    check_name: str
+    asset_key: str
+
+
+class DgApiAssetCheckExecutionList(BaseModel):
+    """List of asset check executions response."""
+
+    items: list[DgApiAssetCheckExecution]
+    cursor: str | None = None
+    has_more: bool = False