[docs] add declarative automation migration guide (#21988)

Internal-RevId: eb7c7f70c6ec914bd78f6c48fad8d90b2fd9009b

Author: cnolanminich

docs/docs/guides/automate/declarative-automation/migrating-from-sensors.md

@@ -0,0 +1,8 @@
+import dagster as dg
+
+
+@dg.asset(
+    deps=["asset_a", "asset_b"],
+    automation_condition=dg.AutomationCondition.on_cron("0 * * * *"),
+)
+def downstream(): ...

examples/docs_snippets/docs_snippets/concepts/declarative_automation/migration/__init__.py

@@ -0,0 +1,16 @@
+import dagster as dg
+
+# start_all_deps_custom
+all_deps_updated_since_target = (
+    dg.AutomationCondition.all_deps_match(
+        dg.AutomationCondition.newly_updated().since(
+            dg.AutomationCondition.asset_matches(
+                "my_downstream_asset",
+                dg.AutomationCondition.newly_requested()
+                | dg.AutomationCondition.newly_updated(),
+            )
+        )
+    )
+    & ~dg.AutomationCondition.in_progress()
+)
+# end_all_deps_custom

examples/docs_snippets/docs_snippets/concepts/declarative_automation/migration/da_all_deps.py

@@ -0,0 +1,8 @@
+import dagster as dg
+
+
+@dg.asset(
+    deps=["asset_a", "asset_b"],
+    automation_condition=dg.AutomationCondition.eager(),
+)
+def downstream(): ...

examples/docs_snippets/docs_snippets/concepts/declarative_automation/migration/da_all_deps_custom.py

@@ -0,0 +1,8 @@
+import dagster as dg
+
+
+@dg.asset(
+    deps=["raw_data"],
+    automation_condition=dg.AutomationCondition.eager(),
+)
+def downstream(): ...

examples/docs_snippets/docs_snippets/concepts/declarative_automation/migration/da_any_dep.py

@@ -0,0 +1,58 @@
+from datetime import datetime, timezone
+
+import dagster as dg
+
+raw_orders = dg.AssetSpec("raw_orders", group_name="external_db")
+raw_customers = dg.AssetSpec("raw_customers", group_name="external_db")
+
+
+@dg.sensor(minimum_interval_seconds=60)
+def db_table_update_sensor(
+    context: dg.SensorEvaluationContext,
+) -> dg.SensorResult:
+    last_checked = float(context.cursor) if context.cursor else 0
+    updated_tables = query_information_schema(last_checked)
+
+    asset_events = [
+        dg.AssetMaterialization(
+            asset_key=table["table_name"],
+            metadata={
+                "last_altered": dg.MetadataValue.text(
+                    datetime.fromtimestamp(
+                        table["last_altered"], tz=timezone.utc
+                    ).isoformat()
+                ),
+            },
+        )
+        for table in updated_tables
+    ]
+
+    new_cursor = (
+        str(max(t["last_altered"] for t in updated_tables))
+        if updated_tables
+        else str(last_checked)
+    )
+
+    return dg.SensorResult(asset_events=asset_events, cursor=new_cursor)
+
+
+@dg.asset(
+    deps=[raw_orders, raw_customers],
+    automation_condition=dg.AutomationCondition.on_cron("0 * * * *"),
+)
+def sales_report(): ...
+
+
+@dg.asset(
+    deps=[raw_orders],
+    automation_condition=dg.AutomationCondition.eager(),
+)
+def order_metrics(): ...
+
+
+def query_information_schema(since_timestamp: float) -> list:
+    # In practice, this would query your database, e.g.:
+    # SELECT table_name, last_altered
+    # FROM information_schema.tables
+    # WHERE last_altered > :since_timestamp
+    ...

examples/docs_snippets/docs_snippets/concepts/declarative_automation/migration/da_basic_trigger.py

@@ -0,0 +1,43 @@
+import dagster as dg
+
+
+def load_sales_data() -> int:
+    return 1500
+
+
+@dg.asset
+def daily_sales() -> dg.MaterializeResult:
+    row_count = load_sales_data()
+    return dg.MaterializeResult(metadata={"row_count": row_count})
+
+
+@dg.asset_check(asset="daily_sales")
+def sales_row_count_check(
+    context: dg.AssetCheckExecutionContext,
+) -> dg.AssetCheckResult:
+    event = context.instance.get_latest_materialization_event(
+        dg.AssetKey("daily_sales")
+    )
+    metadata = event.asset_materialization.metadata
+    row_count = metadata["row_count"].value
+    passed = row_count > 1000
+    return dg.AssetCheckResult(
+        passed=passed,
+        severity=dg.AssetCheckSeverity.ERROR,
+        metadata={"row_count": row_count},
+    )
+
+
+all_deps_checks_passed = dg.AutomationCondition.all_deps_match(
+    dg.AutomationCondition.all_checks_match(
+        ~dg.AutomationCondition.check_failed()
+        | dg.AutomationCondition.will_be_requested(),
+    )
+)
+
+
+@dg.asset(
+    deps=["daily_sales"],
+    automation_condition=dg.AutomationCondition.eager() & all_deps_checks_passed,
+)
+def analytics(): ...

examples/docs_snippets/docs_snippets/concepts/declarative_automation/migration/da_external_db_tables.py

@@ -0,0 +1,29 @@
+import dagster as dg
+
+
+# start_on_missing
+# on_missing: materialize each weekly partition ONCE when all hourly partitions exist.
+# Does NOT re-trigger if hourly partitions re-run later.
+@dg.asset(
+    deps=["hourly_asset"],
+    partitions_def=dg.WeeklyPartitionsDefinition(start_date="2024-01-01"),
+    automation_condition=dg.AutomationCondition.on_missing(),
+)
+def weekly_asset_on_missing(): ...
+
+
+# end_on_missing
+
+
+# start_eager
+# eager: materialize when all hourly partitions exist, AND re-materialize whenever
+# any hourly partition updates. This can produce many redundant downstream runs.
+@dg.asset(
+    deps=["hourly_asset"],
+    partitions_def=dg.WeeklyPartitionsDefinition(start_date="2024-01-01"),
+    automation_condition=dg.AutomationCondition.eager(),
+)
+def weekly_asset_eager(): ...
+
+
+# end_eager

examples/docs_snippets/docs_snippets/concepts/declarative_automation/migration/da_metadata_trigger.py

@@ -0,0 +1,16 @@
+import dagster as dg
+
+
+def process_file(file_path: str): ...
+
+
+@dg.asset(
+    deps=["raw_data"],
+    automation_condition=dg.AutomationCondition.eager(),
+)
+def processed_data(context: dg.AssetExecutionContext):
+    instance = context.instance
+    event = instance.get_latest_materialization_event(dg.AssetKey("raw_data"))
+    metadata = event.asset_materialization.metadata
+    file_path = metadata["file_path"].value
+    return process_file(file_path)