Skip to main content

Customizing on_cron

Ignoring dependencies​

By default, AutomationCondition.on_cron() will wait for all upstream dependencies to be updated before executing the asset it's attached to. In some cases, it can be useful to ignore some upstream dependencies in this calculation. This can be done by passing in an AssetSelection to be ignored:

import dagster as dg

condition = dg.AutomationCondition.on_cron("@hourly").ignore(
    dg.AssetSelection.assets("foo")
)

Alternatively, you can pass in an AssetSelection to be allowed:

import dagster as dg

condition = dg.AutomationCondition.on_cron("@hourly").allow(
    dg.AssetSelection.groups("abc")
)

Waiting for all blocking asset checks to complete before executing​

The AutomationCondition.all_deps_blocking_checks_passed() condition becomes true after all upstream blocking checks have passed.

This can be combined with AutomationCondition.on_cron() to ensure that your asset does not execute if upstream data is failing data quality checks:

import dagster as dg

condition = (
    dg.AutomationCondition.on_cron("@hourly")
    & dg.AutomationCondition.all_deps_blocking_checks_passed()
)

Updating older time partitions​

By default, AutomationCondition.on_cron() will target the latest time partition of an asset.

If you instead want to update partitions on a delay, then you can replace this condition with one that targets a partition that has a specific lag from the latest time window:

from datetime import timedelta

from dagster import AutomationCondition

five_days_ago_condition = AutomationCondition.in_latest_time_window(
    timedelta(days=5)
) & ~AutomationCondition.in_latest_time_window(timedelta(days=4))

condition = AutomationCondition.on_cron("@daily").replace(
    "in_latest_time_window", five_days_ago_condition
)