Skip to content

Commit da8ded6

Browse files
prhaclaude
andcommitted
Add GraphQL schema and resolvers for partitioned asset checks
This PR adds the GraphQL layer for partitioned asset checks: - Add partitionsDefinition field to GraphQL AssetCheck type - Add AssetCheckPartitionStatus type for partition-level status tracking - Add resolver for assetCheckPartitionStatuses query - Update asset graph schema to expose check partition information - Add partition filtering to asset check execution history queries - Generate TypeScript types for UI consumption 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude <noreply@anthropic.com>
1 parent 5dde55b commit da8ded6

8 files changed

Lines changed: 429 additions & 53 deletions

File tree

js_modules/dagster-ui/packages/ui-core/src/graphql/possibleTypes.generated.json

Lines changed: 1 addition & 1 deletion
Some generated files are not rendered by default. Learn more about customizing how changed files appear on GitHub.

js_modules/dagster-ui/packages/ui-core/src/graphql/schema.graphql

Lines changed: 32 additions & 6 deletions
Some generated files are not rendered by default. Learn more about customizing how changed files appear on GitHub.

js_modules/dagster-ui/packages/ui-core/src/graphql/types.ts

Lines changed: 90 additions & 0 deletions
Some generated files are not rendered by default. Learn more about customizing how changed files appear on GitHub.

python_modules/dagster-graphql/dagster_graphql/implementation/fetch_asset_checks.py

Lines changed: 51 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1,22 +1,70 @@
11
from collections.abc import Sequence
2-
from typing import TYPE_CHECKING, Optional
2+
from typing import TYPE_CHECKING, Optional, Union
33

44
from dagster import _check as check
55
from dagster._core.definitions.asset_checks.asset_check_spec import AssetCheckKey
66
from dagster._core.events import ASSET_CHECK_EVENTS, DagsterEventType
77
from dagster._core.loader import LoadingContext
8-
from dagster._core.storage.asset_check_execution_record import AssetCheckExecutionRecordStatus
8+
from dagster._core.remote_representation.handle import RepositoryHandle
9+
from dagster._core.storage.asset_check_execution_record import (
10+
AssetCheckExecutionRecordStatus,
11+
AssetCheckInstanceSupport,
12+
)
913
from dagster._core.storage.dagster_run import RunRecord
14+
from packaging import version
15+
16+
from dagster_graphql.schema.asset_checks import (
17+
GrapheneAssetCheckNeedsAgentUpgradeError,
18+
GrapheneAssetCheckNeedsMigrationError,
19+
GrapheneAssetCheckNeedsUserCodeUpgrade,
20+
)
1021

1122
if TYPE_CHECKING:
1223
from dagster_graphql.schema.asset_checks import GrapheneAssetCheckExecution
1324
from dagster_graphql.schema.entity_key import GrapheneAssetCheckHandle
1425
from dagster_graphql.schema.util import ResolveInfo
1526

1627

28+
def check_asset_checks_support(
29+
graphene_info: "ResolveInfo", repository_handle: RepositoryHandle
30+
) -> Union[
31+
GrapheneAssetCheckNeedsMigrationError,
32+
GrapheneAssetCheckNeedsUserCodeUpgrade,
33+
GrapheneAssetCheckNeedsAgentUpgradeError,
34+
None,
35+
]:
36+
asset_check_support = graphene_info.context.instance.get_asset_check_support()
37+
if asset_check_support == AssetCheckInstanceSupport.NEEDS_MIGRATION:
38+
return GrapheneAssetCheckNeedsMigrationError(
39+
message="Asset checks require an instance migration. Run `dagster instance migrate`."
40+
)
41+
elif asset_check_support == AssetCheckInstanceSupport.NEEDS_AGENT_UPGRADE:
42+
return GrapheneAssetCheckNeedsAgentUpgradeError(
43+
"Asset checks require an agent upgrade to 1.5.0 or greater."
44+
)
45+
else:
46+
check.invariant(
47+
asset_check_support == AssetCheckInstanceSupport.SUPPORTED,
48+
f"Unexpected asset check support status {asset_check_support}",
49+
)
50+
51+
library_versions = graphene_info.context.get_dagster_library_versions(
52+
repository_handle.location_name
53+
)
54+
code_location_version = (library_versions or {}).get("dagster")
55+
if code_location_version and version.parse(code_location_version) < version.parse("1.5"):
56+
return GrapheneAssetCheckNeedsUserCodeUpgrade(
57+
message=(
58+
"Asset checks require dagster>=1.5. Upgrade your dagster"
59+
" version for this code location."
60+
)
61+
)
62+
63+
1764
def fetch_asset_check_executions(
1865
loading_context: LoadingContext,
1966
asset_check_key: AssetCheckKey,
67+
partition: Optional[str],
2068
limit: int,
2169
cursor: Optional[str],
2270
) -> list["GrapheneAssetCheckExecution"]:
@@ -26,6 +74,7 @@ def fetch_asset_check_executions(
2674
check_key=asset_check_key,
2775
limit=limit,
2876
cursor=int(cursor) if cursor else None,
77+
partition=partition,
2978
)
3079

3180
RunRecord.prepare(

0 commit comments

Comments
 (0)