automation conditions that update once per period once all upstreams are materialized, and then again if any upstreams are materialized within that timeframe

automation_asset.py

# runs at 5 minutes past the hour for demonstration purposes.
daily_success_condition = (
    dg.AutomationCondition.newly_updated()
    .since(dg.AutomationCondition.on_cron("*/5 * * * *"))
)

custom_condition = (
    dg.AutomationCondition.on_cron("*/5 * * * *")  # Runs at 9 AM daily
    | (
        dg.AutomationCondition.any_deps_updated()  # When any dependency updates
        & daily_success_condition  # Only if asset was updated since daily cron
        #.since(dg.AutomationCondition.on_cron("* * * * *"))  # Only after the daily cron
        & ~dg.AutomationCondition.any_deps_missing()  # But not if deps are missing
        & ~dg.AutomationCondition.any_deps_in_progress()  # And not if deps are in progress
    )
)

@dg.asset
def upstream_asset_1():
    # Your asset logic here
    pass

@dg.asset
def upstream_asset_2():
    # Your asset logic here
    pass

@dg.asset(automation_condition=custom_condition,
          deps=['upstream_asset_1', 'upstream_asset_2'])
def automation_condition_combination(context: dg.AssetExecutionContext) -> dg.MaterializeResult: ...
def my_asset():
    # Your asset logic here
    pass