This is an automated email from the ASF dual-hosted git repository.
bbovenzi pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new 48c6bb413e0 Warn in UI when the same dag_id is registered from
multiple files (#68791)
48c6bb413e0 is described below
commit 48c6bb413e0aaf7a3ad54e9f5332c56ab13e8af0
Author: Takayoshi Makabe <[email protected]>
AuthorDate: Fri Jul 10 01:12:13 2026 +0900
Warn in UI when the same dag_id is registered from multiple files (#68791)
* Add duplicate dag id warning
* Modify related files
* Add dag_processing.duplicate_dag_id to metrics registry
* Change word (skipped → overwritten)
* Simplified the message
* Delete metrics
* Remove unnecessary import
---------
Co-authored-by: Takayoshi Makabe
<[email protected]>
---
airflow-core/docs/faq.rst | 17 +++++
airflow-core/newsfragments/68791.improvement.rst | 1 +
.../core_api/openapi/v2-rest-api-generated.yaml | 1 +
.../src/airflow/dag_processing/collection.py | 61 ++++++++++++++--
airflow-core/src/airflow/models/dagwarning.py | 1 +
.../airflow/ui/openapi-gen/requests/schemas.gen.ts | 2 +-
.../airflow/ui/openapi-gen/requests/types.gen.ts | 2 +-
.../core_api/routes/public/test_dag_warning.py | 2 +-
.../tests/unit/dag_processing/test_collection.py | 82 ++++++++++++++++++++++
.../src/airflowctl/api/datamodels/generated.py | 1 +
10 files changed, 162 insertions(+), 8 deletions(-)
diff --git a/airflow-core/docs/faq.rst b/airflow-core/docs/faq.rst
index 31e327a1c5a..6860fc72f91 100644
--- a/airflow-core/docs/faq.rst
+++ b/airflow-core/docs/faq.rst
@@ -420,6 +420,23 @@ Additionally, you can catch these issues earlier in your
development workflow by
dynamic values in Dag and Task constructors as part of static linting. See
:ref:`best_practices/code_quality_and_linting` for how to set up ruff with
Airflow-specific rules.
+.. _faq:duplicate-dag-id-warning:
+
+Why do I see a "duplicate dag id" warning?
+------------------------------------------
+
+The Dag processor recognizes each Dag by its ``dag_id``, not by its file path.
If two files define a
+Dag with the same ``dag_id``, only one of them is shown in the UI at a time —
whichever file was parsed
+most recently — and Airflow records a **Dag warning** naming the other,
overwritten file.
+
+This is expected in one common case: you renamed or moved a Dag file. The old
file's row is not removed
+from the database until it goes unobserved for
+:ref:`stale_dag_threshold <config:dag_processor__stale_dag_threshold>`, so the
warning briefly
+appears for both the old and new paths and then clears on its own once the old
file stops being parsed.
+
+If the warning does not clear, two files with the same ``dag_id`` genuinely
coexist. Which one "wins" is
+not guaranteed to stay stable across parses, so rename one of them to give
each Dag a unique ``dag_id``.
+
Dag construction
^^^^^^^^^^^^^^^^
diff --git a/airflow-core/newsfragments/68791.improvement.rst
b/airflow-core/newsfragments/68791.improvement.rst
new file mode 100644
index 00000000000..2c98be3c8ac
--- /dev/null
+++ b/airflow-core/newsfragments/68791.improvement.rst
@@ -0,0 +1 @@
+Warn in the UI when the same dag_id is registered from multiple files.
diff --git
a/airflow-core/src/airflow/api_fastapi/core_api/openapi/v2-rest-api-generated.yaml
b/airflow-core/src/airflow/api_fastapi/core_api/openapi/v2-rest-api-generated.yaml
index 8173fd5fe9c..d3a1d124943 100644
---
a/airflow-core/src/airflow/api_fastapi/core_api/openapi/v2-rest-api-generated.yaml
+++
b/airflow-core/src/airflow/api_fastapi/core_api/openapi/v2-rest-api-generated.yaml
@@ -14570,6 +14570,7 @@ components:
type: string
enum:
- asset conflict
+ - duplicate dag id
- non-existent pool
- runtime varying value
title: DagWarningType
diff --git a/airflow-core/src/airflow/dag_processing/collection.py
b/airflow-core/src/airflow/dag_processing/collection.py
index 9ea66e60ed3..8fa5209ff15 100644
--- a/airflow-core/src/airflow/dag_processing/collection.py
+++ b/airflow-core/src/airflow/dag_processing/collection.py
@@ -31,7 +31,7 @@ import traceback
from typing import TYPE_CHECKING, Any, NamedTuple, TypeVar
import structlog
-from sqlalchemy import delete, func, insert, select, tuple_, update
+from sqlalchemy import delete, false, func, insert, select, tuple_, update
from sqlalchemy.exc import OperationalError
from sqlalchemy.orm import joinedload, load_only
@@ -51,7 +51,7 @@ from airflow.models.asset import (
)
from airflow.models.dag import DagModel, DagOwnerAttributes, DagTag
from airflow.models.dagrun import DagRun
-from airflow.models.dagwarning import DagWarningType
+from airflow.models.dagwarning import DagWarning, DagWarningType
from airflow.models.errors import ParseImportError
from airflow.models.serialized_dag import SerializedDagModel
from airflow.models.trigger import Trigger
@@ -75,7 +75,6 @@ if TYPE_CHECKING:
from sqlalchemy.orm import Session
from sqlalchemy.sql import Select
- from airflow.models.dagwarning import DagWarning
from airflow.models.serialized_dag import DagWriteMetadata
from airflow.typing_compat import Self, Unpack
@@ -325,14 +324,58 @@ def _sync_dag_perms(dag: LazyDeserializedDAG, session:
Session):
security_manager.sync_perm_for_dag(dag_id, dag.access_control)
+def _build_duplicate_dag_id_warnings(
+ dags: Collection[LazyDeserializedDAG],
+ bundle_name: str,
+ session: Session,
+) -> set[DagWarning]:
+ """
+ Detect dag_ids that are defined in multiple files and return DagWarning
objects for each.
+
+ A warning is emitted whenever the incoming file differs from the file
already recorded in
+ DagModel. This covers both accidental duplicates and file renames while
leaving the final
+ interpretation to the user.
+ """
+ dag_by_id = {dag.dag_id: dag for dag in dags}
+ if not dag_by_id:
+ return set()
+
+ existing_rows = session.execute(
+ select(DagModel.dag_id, DagModel.bundle_name,
DagModel.relative_fileloc).where(
+ DagModel.dag_id.in_(dag_by_id),
+ DagModel.is_stale == false(),
+ )
+ )
+
+ warnings: set[DagWarning] = set()
+ for existing in existing_rows:
+ dag = dag_by_id[existing.dag_id]
+ if bundle_name == existing.bundle_name and dag.relative_fileloc ==
existing.relative_fileloc:
+ continue
+
+ message = (
+ f"Also registered from '{existing.relative_fileloc}' (bundle:
'{existing.bundle_name}'), "
+ "which is now overwritten by this file. See the FAQ on duplicate
dag_id warnings for details."
+ )
+ log.warning(
+ "Duplicate dag_id detected",
+ dag_id=dag.dag_id,
+ incoming_fileloc=dag.relative_fileloc,
+ incoming_bundle=bundle_name,
+ existing_fileloc=existing.relative_fileloc,
+ existing_bundle=existing.bundle_name,
+ )
+ warnings.add(DagWarning(dag.dag_id, DagWarningType.DUPLICATE_DAG_ID,
message))
+
+ return warnings
+
+
def _update_dag_warnings(
dag_ids: list[str],
warnings: set[DagWarning],
warning_types: tuple[DagWarningType, ...],
session: Session,
):
- from airflow.models.dagwarning import DagWarning
-
stored_warnings = set(
session.scalars(
select(DagWarning).where(
@@ -446,6 +489,7 @@ def update_dag_parsing_results_in_db(
*,
version_data: dict | None = None,
warning_types: tuple[DagWarningType, ...] = (
+ DagWarningType.DUPLICATE_DAG_ID,
DagWarningType.NONEXISTENT_POOL,
DagWarningType.RUNTIME_VARYING_VALUE,
),
@@ -475,6 +519,13 @@ def update_dag_parsing_results_in_db(
# Retry 'DAG.bulk_write_to_db' & 'SerializedDagModel.bulk_sync_to_db' in
case
# of any Operational Errors
# In case of failures, provide_session handles rollback
+ try:
+ duplicate_warnings = _build_duplicate_dag_id_warnings(dags,
bundle_name, session)
+ except Exception:
+ log.exception("Error building duplicate dag_id warnings.")
+ else:
+ warnings = set(warnings) | duplicate_warnings
+
for attempt in run_with_db_retries(logger=log):
with attempt:
serialize_errors = []
diff --git a/airflow-core/src/airflow/models/dagwarning.py
b/airflow-core/src/airflow/models/dagwarning.py
index d411a3b9386..b0a5a3c93e1 100644
--- a/airflow-core/src/airflow/models/dagwarning.py
+++ b/airflow-core/src/airflow/models/dagwarning.py
@@ -102,5 +102,6 @@ class DagWarningType(str, Enum):
"""
ASSET_CONFLICT = "asset conflict"
+ DUPLICATE_DAG_ID = "duplicate dag id"
NONEXISTENT_POOL = "non-existent pool"
RUNTIME_VARYING_VALUE = "runtime varying value"
diff --git a/airflow-core/src/airflow/ui/openapi-gen/requests/schemas.gen.ts
b/airflow-core/src/airflow/ui/openapi-gen/requests/schemas.gen.ts
index 55b25b2c975..956f4f9c9ae 100644
--- a/airflow-core/src/airflow/ui/openapi-gen/requests/schemas.gen.ts
+++ b/airflow-core/src/airflow/ui/openapi-gen/requests/schemas.gen.ts
@@ -4572,7 +4572,7 @@ export const $DagVersionResponse = {
export const $DagWarningType = {
type: 'string',
- enum: ['asset conflict', 'non-existent pool', 'runtime varying value'],
+ enum: ['asset conflict', 'duplicate dag id', 'non-existent pool', 'runtime
varying value'],
title: 'DagWarningType',
description: `Enum for DAG warning types.
diff --git a/airflow-core/src/airflow/ui/openapi-gen/requests/types.gen.ts
b/airflow-core/src/airflow/ui/openapi-gen/requests/types.gen.ts
index acec153fd1f..85c405ae215 100644
--- a/airflow-core/src/airflow/ui/openapi-gen/requests/types.gen.ts
+++ b/airflow-core/src/airflow/ui/openapi-gen/requests/types.gen.ts
@@ -1216,7 +1216,7 @@ export type DagVersionResponse = {
* This is the set of allowable values for the ``warning_type`` field
* in the DagWarning model.
*/
-export type DagWarningType = 'asset conflict' | 'non-existent pool' | 'runtime
varying value';
+export type DagWarningType = 'asset conflict' | 'duplicate dag id' |
'non-existent pool' | 'runtime varying value';
/**
* Backfill collection serializer for responses in dry-run mode.
diff --git
a/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_dag_warning.py
b/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_dag_warning.py
index 91efee48df7..ade87cccf60 100644
---
a/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_dag_warning.py
+++
b/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_dag_warning.py
@@ -117,5 +117,5 @@ class TestGetDagWarnings:
assert response.status_code == 422
assert (
response_json["detail"][0]["msg"]
- == "Input should be 'asset conflict', 'non-existent pool' or
'runtime varying value'"
+ == "Input should be 'asset conflict', 'duplicate dag id',
'non-existent pool' or 'runtime varying value'"
)
diff --git a/airflow-core/tests/unit/dag_processing/test_collection.py
b/airflow-core/tests/unit/dag_processing/test_collection.py
index ee8d7fa0fcd..ad21350be6b 100644
--- a/airflow-core/tests/unit/dag_processing/test_collection.py
+++ b/airflow-core/tests/unit/dag_processing/test_collection.py
@@ -51,6 +51,7 @@ from airflow.models.asset import (
)
from airflow.models.dag import DagTag
from airflow.models.dagbundle import DagBundleModel
+from airflow.models.dagwarning import DagWarning, DagWarningType
from airflow.models.errors import ParseImportError
from airflow.models.serialized_dag import SerializedDagModel
from airflow.models.trigger import Trigger
@@ -772,6 +773,87 @@ class TestUpdateDagParsingResults:
new_serialized_dags_count =
session.scalar(select(func.count(SerializedDagModel.dag_id)))
assert new_serialized_dags_count == 1
+ @pytest.mark.usefixtures("clean_db")
+ def test_duplicate_dag_id_creates_dag_warning(self, testing_dag_bundle,
session):
+ session.add(
+ DagModel(
+ dag_id="duplicated_dag",
+ bundle_name="testing",
+ fileloc="/opt/airflow/dags/existing.py",
+ relative_fileloc="existing.py",
+ is_stale=False,
+ )
+ )
+ session.flush()
+
+ dag = DAG(dag_id="duplicated_dag")
+ dag.fileloc = "/opt/airflow/dags/current.py"
+ dag.relative_fileloc = "current.py"
+
+ update_dag_parsing_results_in_db(
+ bundle_name="testing",
+ bundle_version=None,
+ dags=[LazyDeserializedDAG.from_dag(dag)],
+ import_errors={},
+ parse_duration=None,
+ warnings=set(),
+ session=session,
+ )
+
+ warning = session.scalar(
+ select(DagWarning).where(
+ DagWarning.dag_id == "duplicated_dag",
+ DagWarning.warning_type == DagWarningType.DUPLICATE_DAG_ID,
+ )
+ )
+
+ assert warning is not None
+ assert "existing.py" in warning.message
+ assert "overwritten" in warning.message
+
+ @pytest.mark.usefixtures("clean_db")
+ def test_duplicate_dag_id_warning_is_removed_when_dag_file_matches(self,
testing_dag_bundle, session):
+ session.add(
+ DagModel(
+ dag_id="same_file_dag",
+ bundle_name="testing",
+ fileloc="/opt/airflow/dags/current.py",
+ relative_fileloc="current.py",
+ is_stale=False,
+ )
+ )
+ session.add(
+ DagWarning(
+ dag_id="same_file_dag",
+ warning_type=DagWarningType.DUPLICATE_DAG_ID,
+ message="Previous duplicate dag_id warning",
+ )
+ )
+ session.flush()
+
+ dag = DAG(dag_id="same_file_dag")
+ dag.fileloc = "/opt/airflow/dags/current.py"
+ dag.relative_fileloc = "current.py"
+
+ update_dag_parsing_results_in_db(
+ bundle_name="testing",
+ bundle_version=None,
+ dags=[LazyDeserializedDAG.from_dag(dag)],
+ import_errors={},
+ parse_duration=None,
+ warnings=set(),
+ session=session,
+ )
+
+ warning = session.scalar(
+ select(DagWarning).where(
+ DagWarning.dag_id == "same_file_dag",
+ DagWarning.warning_type == DagWarningType.DUPLICATE_DAG_ID,
+ )
+ )
+
+ assert warning is None
+
def test_parse_time_written_to_db_on_sync(self, testing_dag_bundle,
session):
"""Test that the parse time is correctly written to the DB after
parsing"""
diff --git a/airflow-ctl/src/airflowctl/api/datamodels/generated.py
b/airflow-ctl/src/airflowctl/api/datamodels/generated.py
index 8ce734890ad..d4e39a8ad19 100644
--- a/airflow-ctl/src/airflowctl/api/datamodels/generated.py
+++ b/airflow-ctl/src/airflowctl/api/datamodels/generated.py
@@ -646,6 +646,7 @@ class DagWarningType(str, Enum):
"""
ASSET_CONFLICT = "asset conflict"
+ DUPLICATE_DAG_ID = "duplicate dag id"
NON_EXISTENT_POOL = "non-existent pool"
RUNTIME_VARYING_VALUE = "runtime varying value"
