Automation Part 2

docs/docs-next/docs/code_examples/guides/automation/asset-sensor-custom-eval.py

@@ -0,0 +1,51 @@
+from dagster import (
+    AssetExecutionContext,
+    AssetKey,
+    AssetMaterialization,
+    Definitions,
+    MaterializeResult,
+    RunRequest,
+    SensorEvaluationContext,
+    SkipReason,
+    asset,
+    asset_sensor,
+    define_asset_job,
+)
+
+
+@asset
+def daily_sales_data(context: AssetExecutionContext):
+    context.log.info("Asset to watch, perhaps some function sets metadata here")
+    yield MaterializeResult(metadata={"specific_property": "value"})
+
+
+@asset
+def weekly_report(context: AssetExecutionContext):
+    context.log.info("Running weekly report")
+
+
+my_job = define_asset_job("my_job", [weekly_report])
+
+
+@asset_sensor(asset_key=AssetKey("daily_sales_data"), job=my_job)
+def daily_sales_data_sensor(context: SensorEvaluationContext, asset_event):
+    # Provide a type hint on the underlying event
+    materialization: AssetMaterialization = (
+        asset_event.dagster_event.event_specific_data.materialization
+    )
+
+    # Example custom logic: Check if the asset metadata has a specific property
+    # highlight-start
+    if "specific_property" in materialization.metadata:
+        context.log.info("Triggering job based on custom evaluation logic")
+        yield RunRequest(run_key=context.cursor)
+    else:
+        yield SkipReason("Asset materialization does not have the required property")
+    # highlight-end
+
+
+defs = Definitions(
+    assets=[daily_sales_data, weekly_report],
+    jobs=[my_job],
+    sensors=[daily_sales_data_sensor],
+)

docs/docs-next/docs/code_examples/guides/automation/asset-sensor-with-config.py

@@ -1,8 +1,11 @@
 from dagster import (
     AssetExecutionContext,
     AssetKey,
+    AssetMaterialization,
+    Config,
     Definitions,
-    EventLogEntry,
+    MaterializeResult,
+    RunConfig,
     RunRequest,
     SensorEvaluationContext,
     asset,
@@ -11,38 +14,49 @@
 )
 
 
+class MyConfig(Config):
+    param1: str
+
+
 @asset
 def daily_sales_data(context: AssetExecutionContext):
     context.log.info("Asset to watch")
+    # highlight-next-line
+    yield MaterializeResult(metadata={"specific_property": "value"})
 
 
 @asset
-def weekly_report(context: AssetExecutionContext):
-    context.log.info("Asset to trigger")
+def weekly_report(context: AssetExecutionContext, config: MyConfig):
+    context.log.info(f"Running weekly report with param1: {config.param1}")
 
 
-my_job = define_asset_job("my_job", [weekly_report])
+my_job = define_asset_job(
+    "my_job",
+    [weekly_report],
+    config=RunConfig(ops={"weekly_report": MyConfig(param1="value")}),
+)
 
 
-# highlight-start
 @asset_sensor(asset_key=AssetKey("daily_sales_data"), job=my_job)
-def daily_sales_data_sensor(context: SensorEvaluationContext, asset_event: EventLogEntry):
-    # This satisifies the type checker. Asset events are guaranteed to have a dagster_event and asset_key.
-    assert asset_event.dagster_event is not None
-    assert asset_event.dagster_event.asset_key is not None
-
-    return RunRequest(
-        run_key=context.cursor,
-        run_config={
-            "ops": {
-                "read_materialization": {
-                    "config": {
-                        "asset_key": asset_event.dagster_event.asset_key.path,
-                    }
+def daily_sales_data_sensor(context: SensorEvaluationContext, asset_event):
+    materialization: AssetMaterialization = (
+        asset_event.dagster_event.event_specific_data.materialization
+    )
+
+    # Example custom logic: Check if the asset metadata has a specific property
+    # highlight-start
+    if "specific_property" in materialization.metadata:
+        yield RunRequest(
+            run_key=context.cursor,
+            run_config=RunConfig(
+                ops={
+                    "weekly_report": MyConfig(
+                        param1=str(materialization.metadata.get("specific_property"))
+                    )
                 }
-            }
-        },
-    )  # highlight-end
+            ),
+        )
+    # highlight-end
 
 
 defs = Definitions(

docs/docs-next/docs/code_examples/guides/automation/multi-asset-sensor.py

@@ -0,0 +1,32 @@
+from dagster import (
+    AssetKey,
+    MultiAssetSensorEvaluationContext,
+    RunRequest,
+    asset,
+    define_asset_job,
+    multi_asset_sensor,
+)
+
+
+@asset
+def target_asset():
+    pass
+
+
+downstream_job = define_asset_job("downstream_job", [target_asset])
+
+
+@multi_asset_sensor(
+    monitored_assets=[
+        AssetKey("upstream_asset_1"),
+        AssetKey("upstream_asset_2"),
+    ],
+    job=downstream_job,
+)
+def my_multi_asset_sensor(context: MultiAssetSensorEvaluationContext):
+    run_requests = []
+    for asset_key, materialization in context.latest_materialization_records_by_key().items():
+        if materialization:
+            run_requests.append(RunRequest(asset_selection=[asset_key]))
+            context.advance_cursor({asset_key: materialization})
+    return run_requests

docs/docs-next/docs/code_examples/guides/automation/schedule-with-partition.py

@@ -0,0 +1,22 @@
+from dagster import (
+    DailyPartitionsDefinition,
+    asset,
+    build_schedule_from_partitioned_job,
+    define_asset_job,
+)
+
+daily_partition = DailyPartitionsDefinition(start_date="2024-05-20")
+
+
+@asset(partitions_def=daily_partition)
+def daily_asset(): ...
+
+
+partitioned_asset_job = define_asset_job("partitioned_job", selection=[daily_asset])
+
+# highlight-start
+# This partition will run daily
+asset_partitioned_schedule = build_schedule_from_partitioned_job(
+    partitioned_asset_job,
+)
+# highlight-end

docs/docs-next/docs/concepts/automation.md

@@ -1 +1,66 @@
-# Automation
+---
+title: About Automation
+---
+
+Dagster has support for several types of automation. All automation in Dagster responds to some external event.
+
+The first system, and the most basic, is the [Schedule](/guides/automation/schedules), which responds to time.
+
+[Sensors](/guides/automation/sensors) are like schedules, but they respond to an external event defined by the user.
+
+[Asset Sensors](/guides/automation/asset-sensors) are a special case of sensor that responds to changes in asset materialization
+as reported by the Event Log.
+
+Finally, the Declarative Automation system is a
+more complex system that uses conditions on the assets to determine when to execute.
+
+## About Schedules
+
+In Dagster, a schedule is defined by the `ScheduleDefinition` class, or through the `@schedule` decorator. The `@schedule`
+decorator is more flexible than the `ScheduleDefinition` class, allowing you to configure job behaviour or emit log messages
+as the schedule is processed.
+
+Schedules were one of the first types of automation in Dagster, created before the introduction of Software-Defined Assets.
+As such, you may find that many of the examples can seem foreign if you are used to only working within the asset framework.
+
+For more on how assets and ops inter-relate, read about [Assets and Ops](/concepts/assets#assets-and-ops)
+
+The `dagster-daemon` process is responsible for submitting runs by checking each schedule at a regular interval to determine
+if it's time to execute the underlying job.
+
+A schedule can be thought of as a wrapper around two pieces:
+
+- A `JobDefinition`, which is a set of assets to materialize or ops to execute.
+- A `cron` string, which describes the schedule.
+
+### Defining a schedule using `ScheduleDefinition`
+
+```python
+ecommerce_schedule = ScheduleDefinition(
+    job=ecommerce_job,
+    cron_schedule="15 5 * * 1-5",
+)
+```
+
+By default, schedules aren't enabled. You can enable them by visiting the Automation tab and toggling the schedule,
+or set a default status to `RUNNING` when you define the schedule.
+
+```python
+ecommerce_schedule = ScheduleDefinition(
+    job=ecommerce_job,
+    cron_schedule="15 5 * * 1-5",
+    default_status=DefaultScheduleStatus.RUNNING,
+)
+```
+
+### Defining a schedule using `@schedule`
+
+If you want more control over the schedule, you can use the `@schedule` decorator. In doing so, you are then responsible for either
+emitting a `RunRequest` or a `SkipReason`. You can also emit logs, which will be visible in the Dagster UI for a given schedule's tick history.
+
+```python
+@schedule(cron_schedule="15 5 * * 1-5")
+def ecommerce_schedule(context):
+    context.log.info("This log message will be visible in the Dagster UI.")
+    return RunRequest()
+```

docs/docs-next/docs/concepts/automation/schedules.md

@@ -1,6 +0,0 @@
----
-title: "Schedules"
-sidebar_position: 10
----
-
-# Schedules

docs/docs-next/docs/guides/automation.md

@@ -35,8 +35,8 @@ Dagster offers several ways to automate pipeline execution:
 ## Schedules
 
 Schedules allow you to run jobs at specified times, like "every Monday at 9 AM" or "daily at midnight."
-A schedule combines a selection of assets, known as a [Job](/concepts/ops-jobs), and a [cron expression](https://en.wikipedia.org/wiki/Cron) 
-in order to define when the job should be run.
+A schedule combines a selection of assets, known as a [Job](/concepts/ops-jobs), and a [cron expression](https://en.wikipedia.org/wiki/Cron)
+to define when the job should be run.
 
 To make creating cron expressions easier, you can use an online tool like [Crontab Guru](https://crontab.guru/).
 
@@ -51,36 +51,38 @@ For more information about how Schedules work, see [About Schedules](/concepts/s
 
 ## Sensors
 
-Sensors allow you to trigger runs based on events or conditions, like a new file arriving or an external system status change.
+Sensors allow you to trigger runs based on events or conditions that you define, like a new file arriving or an external system status change.
 
-Like schedules, sensors operate on a selection of assets, known as [Jobs](/concepts/ops-jobs) and can either start a pipeline 
-through a Run or log a reason for not starting a pipeline using a SkipReason.
-
-However, unlike schedules, sensors are triggered by events that you define.
 You must provide a function that the sensor will use to determine if it should trigger a run.
 
-### When to use Sensors
+Like schedules, sensors operate on a selection of assets, known as [Jobs](/concepts/ops-jobs) and can either start a pipeline through a Run or log a reason for not starting a pipeline using a SkipReason.
+
+
+### When to use sensors
 
 - You need event-driven automation
 - You want to react to changes in external systems
 
 For more examples of how to create sensors, see the [How-To Use Sensors](/guides/automation/sensors) guide.
 
-For more information about how Sensors work, see the [About Sensors](/concepts/sensors) concept page.
+For more information about how sensors work, see the [About Sensors](/concepts/sensors) concept page.
 
 ## Asset sensors
 
 Asset Sensors trigger jobs when specified assets are materialized, allowing you to create dependencies between jobs or code locations.
 
-### When to use Asset Sensors
+### When to use Asset sensors
 
 - You need to trigger jobs based on asset materializations
 - You want to create dependencies between different jobs or code locations
 
 For more examples of how to create asset sensors, see the [How-To Use Asset Sensors](/guides/automation/asset-sensors) guide.
 
+## Declarative automation
+
+TODO: Add content
 
-## Choosing the Right Automation Method
+## How to choose the right automation method
 
 Consider these factors when selecting an automation method:
 
@@ -91,17 +93,17 @@ Consider these factors when selecting an automation method:
 
 Use this table to help guide your decision:
 
-| Method | Best For | Works With |
-|--------|----------|------------|
-| Schedules | Regular, time-based job runs | Assets, Ops, Graphs |
-| Sensors | Event-driven automation | Assets, Ops, Graphs |
-| Declarative Automation | Asset-centric, condition-based updates | Assets only |
-| Asset Sensors | Cross-job/location asset dependencies | Assets only |
+| Method                 | Best For                               | Works With          |
+| ---------------------- | -------------------------------------- | ------------------- |
+| Schedules              | Regular, time-based job runs           | Assets, Ops, Graphs |
+| Sensors                | Event-driven automation                | Assets, Ops, Graphs |
+| Declarative Automation | Asset-centric, condition-based updates | Assets only         |
+| Asset Sensors          | Cross-job/location asset dependencies  | Assets only         |
 
-## Next Steps
+## Next steps
 
 - Learn more about [advanced scheduling patterns] - TODO ADD LINK
 - Explore [complex sensor examples] - TODO ADD LINK
 - Dive into [Declarative Automation best practices] - TODO ADD LINK
 
-By understanding and effectively using these automation methods, you can build more efficient data pipelines that respond to your specific needs and constraints.
+By understanding and effectively using these automation methods, you can build more efficient data pipelines that respond to your specific needs and constraints.