Better scheduling across time grains

[sryza]

It's common for an asset that gets updated daily to depend on an asset that gets updated hourly. Or vice versa. Currently, expressing and monitoring this kind of execution pattern is awkward:

@asset
def asset1():
    ...

@asset
def asset2(asset1):
    ...

@schedule(
    cron_schedule="@hourly",
    job=define_asset_job("hourly_job", selection=AssetSelection.assets(asset1)),
)
def hourly_schedule(context):
    return RunRequest(run_key=None, tags={"hour_of_day": context.scheduled_execution_time.hour})

@asset_sensor(
    asset_key=asset1.key,
    job=define_asset_job("daily_job", selection=AssetSelection.assets(asset2)),
)
def daily_downstream_sensor(context, event):
    upstream_run_tags = context.instance.get_run_by_id(event.run_id).tags
    if upstream_run_tags["hour_of_day"] == 5:
        return RunRequest(run_key=None)


@repository
def repo():
    return [asset1, asset2, hourly_schedule, daily_downstream_sensor]

Ideally, there would be a way to express this more cleanly.