Add AssetAndTimeSchedule timetable

[nailo2c]

related: #58056

Why

Use case: #58056

How

1. Implement AssetAndTimeSchedule class

Basically almost the same as AssetOrTimeSchedule, but this class inherits from Timetable and delegates run_id generation to the wrapped timetable.

2. Assign AssetAndTimeSchedule to the non_asset_dags set

Modifying ‎DagModel.dags_needing_dagruns‎ to prevent AssetAndTimeSchedule from being assigned to triggered_date_by_dag.

3. Add starting logic for AssetAndTimeSchedule in the scheduler.

Updating the DAG status from QUEUED to RUNNING when both the timetable and asset conditions are meet.

What

Here are my test DAGs:

upstream_asset_producer

import pendulum
from airflow.models.dag import DAG
from airflow.operators.bash import BashOperator
from airflow.sdk.definitions.asset import Asset

my_asset = Asset("/my/example/asset")

with DAG(
    dag_id="upstream_asset_producer",
    start_date=pendulum.datetime(2025, 1, 1, tz="UTC"),
    schedule="0 * * * *",
    catchup=False,
    tags=["example", "upstream"],
) as upstream_dag:
    BashOperator(
        task_id="produce_asset",
        bash_command="echo 'Asset produced!'",
        outlets=[my_asset],
    )

downstream_asset_and_time_consumer

import pendulum
from airflow.models.dag import DAG
from airflow.operators.bash import BashOperator
from airflow.sdk.definitions.asset import Asset
from airflow.timetables.trigger import CronTriggerTimetable
from airflow.timetables.assets import AssetAndTimeSchedule

my_asset = Asset("/my/example/asset")

with DAG(
    dag_id="downstream_asset_and_time_consumer",
    start_date=pendulum.datetime(2025, 1, 1, tz="UTC"),
    schedule=AssetAndTimeSchedule(
        timetable=CronTriggerTimetable("4 * * * *", timezone="UTC"),
        assets=[my_asset],
    ),
    catchup=False,
    max_active_runs=1,
    tags=["example", "downstream", "asset-and-time"],
) as downstream_dag:
    BashOperator(
        task_id="consume_asset",
        bash_command="echo 'Asset condition and time condition were both met!'",
    )

From the screenshots, the downstream DAG runs as expected: it runs only after both the timetable and the assets are ready.

1. Upstream dag

   

2. Asset production time

   

3. Downstream, AssetAndTimeSchedule

   

Discussion

Since I added this logic in dag.py to guarantee that only AssetTriggeredTimetable can be added to triggered_date_by_dag.

if not isinstance(timetable, AssetTriggeredTimetable):
    del adrq_by_dag[dag_id]
    continue

Should I remove the extra guard logic in SchedulerJobRunner._create_dag_runs_asset_triggered?

airflow/airflow-core/src/airflow/jobs/scheduler_job_runner.py

Lines 1800 to 1805 in 9e7b36e

	if not isinstance(dag.timetable, AssetTriggeredTimetable):
	self.log.error(
	"DAG '%s' was asset-scheduled, but didn't have an AssetTriggeredTimetable!",
	dag_model.dag_id,
	)
	continue

[nailo2c]

Copilot's feedback are helpful, I've fixed both issues.

1. Thanks! yep the permalink was probably the thing that made me confused 🤦‍♂️

I've updated the link in the PR description to a permalink.

3. I'm thinking about whether we have a way to reduce the confusion. If I were to be the user, I might be confused. 🤔 Do you think there's an easy way to change how the schedule box displays based on the current status? (I "guess" not, but worth trying)

I think that would require significant changes across the API and frontend. I'd prefer to keep this PR focused on the core functionality and address the UI improvements in a follow-up if needed.

[Lee-W]

mostly good, 2 questions

[Lee-W]

one last question, but mostly good

[Lee-W]

It's not a feature I would ever use, but I'm good with it, and it seems there's an actual use case. Would like to know how @uranusjr thinks. Thanks!

[potiuk]

@uranusjr — Lee-W asked for your take on this 25 days ago. When you have a moment, could you weigh in? It's the only thing blocking a decision on the PR right now. Thanks!

---

Note: This comment was drafted by an AI-assisted triage tool and may contain mistakes.

---

Drafted-by: Claude Code (Opus 4.7); reviewed by @potiuk before posting

[Lee-W]

It's a less common but reasonable use case. so if we have another committer's good with this feature, I'm good with it as well :)

[potiuk]

@nailo2c — This PR is labeled ready for maintainer review, and a maintainer left feedback about 7 days ago that is still awaiting your response. Could you take a look and either push an update or reply in the thread? Once you have responded, a maintainer will pick it back up from the review queue.

No rush — just flagging it so the PR does not stall.

---

Note: This comment was drafted by an AI-assisted triage tool and may contain mistakes. Once you have addressed the points above, an Apache Airflow maintainer — a real person — will take the next look at your PR. We use this two-stage triage process so that our maintainers' limited time is spent where it matters most: the conversation with you.

[nailo2c]

Thanks for the reminder. I believe the previous review comments have been addressed and all review threads are resolved. Based on the latest discussion, it seems the remaining point is waiting for another committer's opinion on the feature. Please let me know if there is anything else I should update :)