ashb commented on code in PR #61833:
URL: https://github.com/apache/airflow/pull/61833#discussion_r2812945393
##########
airflow-core/src/airflow/models/backfill.py:
##########
@@ -100,6 +100,14 @@ class InvalidBackfillDate(AirflowException):
"""
+class DeniedDagRunType(AirflowException):
Review Comment:
This is nothing to do with backfills, so this should be elsewhere
##########
airflow-core/src/airflow/migrations/versions/0104_3_2_0_add_allowed_run_types_to_dag.py:
##########
@@ -0,0 +1,50 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+"""
+Add allowed_run_types to dag.
+
+Revision ID: e42d9fcd10d9
+Revises: f8c9d7e6b5a4
+Create Date: 2026-02-12 11:49:40.753440
+
+"""
+
+from __future__ import annotations
+
+import sqlalchemy as sa
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision = "e42d9fcd10d9"
+down_revision = "f8c9d7e6b5a4"
+branch_labels = None
+depends_on = None
+airflow_version = "3.2.0"
+
+
+def upgrade():
+ """Add allowed_run_types column to dag table."""
+ with op.batch_alter_table("dag", schema=None) as batch_op:
+ batch_op.add_column(sa.Column("allowed_run_types", sa.JSON(),
nullable=True))
Review Comment:
Json is overkill/allows more than we want here - maybe for postgres this
would be better as the native ARRAY type?
##########
airflow-core/src/airflow/models/backfill.py:
##########
@@ -502,6 +510,14 @@ def _create_backfill(
if not serdag:
raise DagNotFound(f"Could not find dag {dag_id}")
+ dag_model = session.scalar(select(DagModel).where(DagModel.dag_id ==
dag_id).limit(1))
Review Comment:
Is there any way we could combine this and the next select on to one?
##########
airflow-core/src/airflow/dag_processing/collection.py:
##########
@@ -600,6 +601,14 @@ def update_dags(
dm.timetable_description = dag.timetable.description
dm.fail_fast = dag.fail_fast if dag.fail_fast is not None else
False
+ allowed_types = dag.allowed_run_types
+ if allowed_types:
+ dm.allowed_run_types = sorted(
+ v.value if isinstance(v, enum.Enum) else v for v in
allowed_types
+ )
Review Comment:
This looks a bit odd. What stop us doing
```suggestion
dm.allowed_run_types = sorted(allowed_types)
```
##########
airflow-core/src/airflow/jobs/scheduler_job_runner.py:
##########
@@ -2060,6 +2082,16 @@ def _create_dag_runs_asset_triggered(
)
continue
+ if (
+ dag_model.allowed_run_types is not None
+ and DagRunType.ASSET_TRIGGERED.value not in
dag_model.allowed_run_types
+ ):
+ self.log.warning(
+ "Dag does not allow asset-triggered runs; skipping",
Review Comment:
This shouldn't be a skip here - i think this should be a parse time error
when you try to reference a dag from an asset without allowing the type
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
