Assets, resources definitions and checks

.tmp_dagster_home_e8m4zq2a/storage/442543c7-de52-4f6f-8d70-03436a4fa5c6/compute_logs/knnclhpc.err

@@ -0,0 +1,5 @@
+2026-07-23 07:53:47 +0100 - dagster - DEBUG - __ASSET_JOB - 442543c7-de52-4f6f-8d70-03436a4fa5c6 - 130644 - LOGS_CAPTURED - Capturing logs for process (pid: 130644).
+2026-07-23 07:53:47 +0100 - dagster - DEBUG - __ASSET_JOB - 442543c7-de52-4f6f-8d70-03436a4fa5c6 - 130644 - orders_aggregation_orders_aggregation_check - STEP_START - Started execution of step "orders_aggregation_orders_aggregation_check".
+2026-07-23 07:53:47 +0100 - dagster - DEBUG - __ASSET_JOB - 442543c7-de52-4f6f-8d70-03436a4fa5c6 - 130644 - orders_aggregation_orders_aggregation_check - STEP_OUTPUT - Yielded output "result" of type "Any". (Type check passed).
+2026-07-23 07:53:47 +0100 - dagster - DEBUG - __ASSET_JOB - 442543c7-de52-4f6f-8d70-03436a4fa5c6 - 130644 - orders_aggregation_orders_aggregation_check - ASSET_CHECK_EVALUATION - Asset check 'orders_aggregation_check' on 'orders_aggregation' passed.
+2026-07-23 07:53:47 +0100 - dagster - DEBUG - __ASSET_JOB - 442543c7-de52-4f6f-8d70-03436a4fa5c6 - 130644 - orders_aggregation_orders_aggregation_check - STEP_SUCCESS - Finished execution of step "orders_aggregation_orders_aggregation_check" in 34ms.

src/dagster_intro/defs/assets.py

@@ -76,3 +76,19 @@ def orders_aggregation(duckdb: DuckDBResource):
             );
             """
         )
+
+
+@dg.asset_check(asset="orders_aggregation")
+def orders_aggregation_check(duckdb: DuckDBResource) -> dg.AssetCheckResult:
+    table_name = "orders_aggregation"
+    with duckdb.get_connection() as conn:
+        row_count = conn.execute(f"select count(*) from {table_name}").fetchone()[0]
+
+    if row_count == 0:
+        return dg.AssetCheckResult(
+            passed=False, metadata={"message": "Order aggregation check failed"}
+        )
+
+    return dg.AssetCheckResult(
+        passed=True, metadata={"message": "Order aggregation check passed"}
+    )