[demo] Passing in the LoadingContext through the blocking batch load fns

python_modules/dagster-graphql/dagster_graphql/schema/asset_checks.py

@@ -158,7 +158,7 @@ async def resolve_executionForLatestMaterialization(
         record = await AssetCheckExecutionRecord.gen(graphene_info.context, self._asset_check.key)
         return (
             GrapheneAssetCheckExecution(record)
-            if record and record.targets_latest_materialization(graphene_info.context)
+            if record and await record.targets_latest_materialization(graphene_info.context)
             else None
         )
 

python_modules/dagster-test/dagster_test/toys/auto_materializing/large_graph.py

@@ -2,58 +2,83 @@
 from typing import List, NamedTuple, Optional, Sequence
 
 from dagster import (
+    AssetCheckResult,
+    AssetCheckSpec,
+    AssetKey,
     AssetsDefinition,
     AutoMaterializePolicy,
+    AutomationCondition,
     DailyPartitionsDefinition,
     HourlyPartitionsDefinition,
+    MaterializeResult,
     PartitionsDefinition,
     StaticPartitionsDefinition,
     asset,
     repository,
 )
+from dagster._core.execution.context.asset_execution_context import AssetExecutionContext
+from dagster._utils.warnings import disable_dagster_warnings
 
 
 class AssetLayerConfig(NamedTuple):
     n_assets: int
     n_upstreams_per_asset: int = 0
     partitions_def: Optional[PartitionsDefinition] = None
+    n_checks_per_asset: int = 0
 
 
 def build_assets(
     id: str,
     layer_configs: Sequence[AssetLayerConfig],
-    auto_materialize_policy: AutoMaterializePolicy = AutoMaterializePolicy.eager(),
+    automation_condition: AutomationCondition = AutomationCondition.eager(),
 ) -> List[AssetsDefinition]:
-    layers = []
-
-    for layer_config in layer_configs:
-        parent_index = 0
-        layer = []
-        for i in range(layer_config.n_assets):
-            if layer_config.n_upstreams_per_asset > 0:
-                # each asset connects to n_upstreams_per_asset assets from the above layer, chosen
-                # in a round-robin manner
-                non_argument_deps = {
-                    layers[-1][(parent_index + j) % len(layers[-1])].key
-                    for j in range(layer_config.n_upstreams_per_asset)
-                }
-                parent_index += layer_config.n_upstreams_per_asset
-            else:
-                non_argument_deps = set()
-
-            @asset(
-                partitions_def=layer_config.partitions_def,
-                name=f"{id}_{len(layers)}_{i}",
-                auto_materialize_policy=auto_materialize_policy,
-                non_argument_deps=non_argument_deps,
-            )
-            def _asset():
-                pass
-
-            layer.append(_asset)
-        layers.append(layer)
-
-    return list(itertools.chain(*layers))
+    with disable_dagster_warnings():
+        layers = []
+
+        for layer_config in layer_configs:
+            parent_index = 0
+            layer = []
+            for i in range(layer_config.n_assets):
+                if layer_config.n_upstreams_per_asset > 0:
+                    # each asset connects to n_upstreams_per_asset assets from the above layer, chosen
+                    # in a round-robin manner
+                    non_argument_deps = {
+                        layers[-1][(parent_index + j) % len(layers[-1])].key
+                        for j in range(layer_config.n_upstreams_per_asset)
+                    }
+                    parent_index += layer_config.n_upstreams_per_asset
+                else:
+                    non_argument_deps = set()
+
+                name = f"{id}_{len(layers)}_{i}"
+
+                @asset(
+                    partitions_def=layer_config.partitions_def,
+                    name=name,
+                    automation_condition=automation_condition,
+                    non_argument_deps=non_argument_deps,
+                    check_specs=[
+                        AssetCheckSpec(
+                            name=f"check{k}",
+                            asset=AssetKey(name),
+                            automation_condition=automation_condition,
+                        )
+                        for k in range(layer_config.n_checks_per_asset)
+                    ],
+                )
+                def _asset(context: AssetExecutionContext) -> MaterializeResult:
+                    return MaterializeResult(
+                        asset_key=context.asset_key,
+                        check_results=[
+                            AssetCheckResult(check_name=key.name, passed=True)
+                            for key in context.selected_asset_check_keys
+                        ],
+                    )
+
+                layer.append(_asset)
+            layers.append(layer)
+
+        return list(itertools.chain(*layers))
 
 
 hourly = HourlyPartitionsDefinition(start_date="2022-01-01-00:00")
@@ -74,6 +99,7 @@ def auto_materialize_large_time_graph():
             AssetLayerConfig(n_assets=100, n_upstreams_per_asset=4, partitions_def=daily),
             AssetLayerConfig(n_assets=100, n_upstreams_per_asset=4, partitions_def=daily),
         ],
+        automation_condition=AutoMaterializePolicy.eager().to_automation_condition(),
     )
 
 
@@ -88,4 +114,5 @@ def auto_materialize_large_static_graph():
             AssetLayerConfig(n_assets=100, n_upstreams_per_asset=4, partitions_def=static),
             AssetLayerConfig(n_assets=100, n_upstreams_per_asset=4, partitions_def=None),
         ],
+        automation_condition=AutoMaterializePolicy.eager().to_automation_condition(),
     )