Adjust warnings, import errors, and Dag source models for sdk/importers

airflow-core/docs/migrations-ref.rst

@@ -39,7 +39,11 @@ Here's the list of all the Database Migrations that are executed via when you ru
 +-------------------------+------------------+-------------------+--------------------------------------------------------------+
 | Revision ID             | Revises ID       | Airflow Version   | Description                                                  |
 +=========================+==================+===================+==============================================================+
-| ``a61f0c9d2b47`` (head) | ``c9f4b3e7a218`` | ``3.4.0``         | Allocate attempt numbers for pending retries and cleared     |
+| ``e5a91c7f42b3`` (head) | ``ca8499dc1004`` | ``3.4.0``         | Add language column to dag_code.                             |
++-------------------------+------------------+-------------------+--------------------------------------------------------------+
+| ``ca8499dc1004``        | ``a61f0c9d2b47`` | ``3.4.0``         | Add source_reference to import_error.                        |
++-------------------------+------------------+-------------------+--------------------------------------------------------------+
+| ``a61f0c9d2b47``        | ``c9f4b3e7a218`` | ``3.4.0``         | Allocate attempt numbers for pending retries and cleared     |
 |                         |                  |                   | tasks.                                                       |
 +-------------------------+------------------+-------------------+--------------------------------------------------------------+
 | ``c9f4b3e7a218``        | ``3b7a91c5df20`` | ``3.4.0``         | Replace the ``job.team_name`` column with a ``job_team``     |

airflow-core/src/airflow/api/common/delete_dag.py

@@ -22,7 +22,7 @@
 import logging
 from typing import TYPE_CHECKING, cast
 
-from sqlalchemy import delete, select
+from sqlalchemy import delete, or_, select
 
 from airflow import models
 from airflow.exceptions import AirflowException, DagNotFound
@@ -82,7 +82,10 @@ def delete_dag(dag_id: str, keep_records_in_log: bool = True, *, session: Sessio
     # This handles the case when the dag_id is changed in the file
     session.execute(
         delete(ParseImportError).where(
-            ParseImportError.filename == dag.relative_fileloc,
+            or_(
+                ParseImportError.source_reference == dag.relative_fileloc,
+                ParseImportError.filename == dag.relative_fileloc,
+            ),
             ParseImportError.bundle_name == dag.bundle_name,
         )
     )

airflow-core/src/airflow/api_fastapi/core_api/datamodels/dag_sources.py

@@ -28,3 +28,4 @@ class DAGSourceResponse(BaseModel):
     dag_id: str
     version_number: int | None
     dag_display_name: str = Field(validation_alias=AliasPath("dag_model", "dag_display_name"))
+    language: str | None = None

airflow-core/src/airflow/api_fastapi/core_api/datamodels/dag_warning.py

@@ -23,14 +23,14 @@
 from pydantic import AliasPath, Field
 
 from airflow.api_fastapi.core_api.base import BaseModel
-from airflow.models.dagwarning import DagWarningType
+from airflow.models.dagwarning import DagWarningTypeValue
 
 
 class DAGWarningResponse(BaseModel):
     """Dag Warning serializer for responses."""
 
     dag_id: str
-    warning_type: DagWarningType
+    warning_type: DagWarningTypeValue
     message: str
     timestamp: datetime
     dag_display_name: str = Field(validation_alias=AliasPath("dag_model", "dag_display_name"))

airflow-core/src/airflow/api_fastapi/core_api/datamodels/import_error.py

@@ -31,6 +31,7 @@ class ImportErrorResponse(BaseModel):
     id: int = Field(alias="import_error_id")
     timestamp: datetime
     filename: str
+    source_reference: str | None
     bundle_name: str | None
     stacktrace: str = Field(alias="stack_trace")
 

airflow-core/src/airflow/api_fastapi/core_api/openapi/v2-rest-api-generated.yaml

@@ -3805,6 +3805,9 @@ paths:
         schema:
           anyOf:
           - $ref: '#/components/schemas/DagWarningType'
+          - type: string
+            maxLength: 50
+            pattern: ^[a-z][a-z0-9_]*:[a-z0-9_.\-]+$
           - type: 'null'
           title: Warning Type
       - name: limit
@@ -5256,13 +5259,13 @@ paths:
             type: string
           description: 'Attributes to order by, multi criteria sort is supported.
             Prefix with `-` for descending order. Supported attributes: `id, timestamp,
-            filename, bundle_name, stacktrace, import_error_id`'
+            filename, source_reference, bundle_name, stacktrace, import_error_id`'
           default:
           - id
           title: Order By
         description: 'Attributes to order by, multi criteria sort is supported. Prefix
           with `-` for descending order. Supported attributes: `id, timestamp, filename,
-          bundle_name, stacktrace, import_error_id`'
+          source_reference, bundle_name, stacktrace, import_error_id`'
       - name: filename_pattern
         in: query
         required: false
@@ -14283,6 +14286,11 @@ components:
         dag_display_name:
           type: string
           title: Dag Display Name
+        language:
+          anyOf:
+          - type: string
+          - type: 'null'
+          title: Language
       type: object
       required:
       - content
@@ -14345,7 +14353,12 @@ components:
           type: string
           title: Dag Id
         warning_type:
-          $ref: '#/components/schemas/DagWarningType'
+          anyOf:
+          - $ref: '#/components/schemas/DagWarningType'
+          - type: string
+            maxLength: 50
+            pattern: ^[a-z][a-z0-9_]*:[a-z0-9_.\-]+$
+          title: Warning Type
         message:
           type: string
           title: Message
@@ -15450,6 +15463,11 @@ components:
         filename:
           type: string
           title: Filename
+        source_reference:
+          anyOf:
+          - type: string
+          - type: 'null'
+          title: Source Reference
         bundle_name:
           anyOf:
           - type: string
@@ -15469,6 +15487,7 @@ components:
       - import_error_id
       - timestamp
       - filename
+      - source_reference
       - bundle_name
       - stack_trace
       - file_token

airflow-core/src/airflow/api_fastapi/core_api/routes/public/dag_sources.py

@@ -103,6 +103,7 @@ def get_dag_source(
         content=content,
         version_number=dag_version.version_number,
         dag_display_name=dag_version.dag_model.dag_display_name,
+        language=dag_version.dag_code.language,
     )
 
     if accept == Mimetype.TEXT:

airflow-core/src/airflow/api_fastapi/core_api/routes/public/dag_warning.py

@@ -40,7 +40,7 @@
     DAGWarningCollectionResponse,
 )
 from airflow.api_fastapi.core_api.security import ReadableDagWarningsFilterDep, requires_access_dag
-from airflow.models.dagwarning import DagWarning, DagWarningType
+from airflow.models.dagwarning import DagWarning, DagWarningTypeValue
 
 dag_warning_router = AirflowRouter(tags=["DagWarning"])
 
@@ -52,8 +52,8 @@
 def list_dag_warnings(
     dag_id: Annotated[FilterParam[str | None], Depends(filter_param_factory(DagWarning.dag_id, str | None))],
     warning_type: Annotated[
-        FilterParam[DagWarningType | None],
-        Depends(filter_param_factory(DagWarning.warning_type, DagWarningType | None)),
+        FilterParam[DagWarningTypeValue | None],
+        Depends(filter_param_factory(DagWarning.warning_type, DagWarningTypeValue | None)),
     ],
     limit: QueryLimit,
     offset: QueryOffset,

airflow-core/src/airflow/api_fastapi/core_api/routes/public/import_error.py

@@ -147,6 +147,7 @@ def get_import_errors(
                     "id",
                     "timestamp",
                     "filename",
+                    "source_reference",
                     "bundle_name",
                     "stacktrace",
                 ],
@@ -263,6 +264,7 @@ def get_import_errors(
             ParseImportError.id,
             ParseImportError.timestamp,
             ParseImportError.filename,
+            ParseImportError.source_reference,
             ParseImportError.bundle_name,
             ParseImportError.stacktrace,
         )

airflow-core/src/airflow/dag_processing/collection.py

@@ -424,6 +424,7 @@ def _update_import_errors(
                 )
                 .values(
                     filename=relative_fileloc,
+                    source_reference=relative_fileloc,
                     bundle_name=bundle_name_,
                     timestamp=utcnow(),
                     stacktrace=stacktrace,
@@ -447,6 +448,7 @@ def _update_import_errors(
         else:
             import_error = ParseImportError(
                 filename=relative_fileloc,
+                source_reference=relative_fileloc,
                 bundle_name=bundle_name_,
                 timestamp=utcnow(),
                 stacktrace=stacktrace,

airflow-core/src/airflow/migrations/versions/0139_3_4_0_add_source_reference_to_import_error.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 source_reference to import_error.
+
+Revision ID: ca8499dc1004
+Revises: a61f0c9d2b47
+Create Date: 2026-08-27 12:45:02.276898
+
+"""
+
+from __future__ import annotations
+
+import sqlalchemy as sa
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision = "ca8499dc1004"
+down_revision = "a61f0c9d2b47"
+branch_labels = None
+depends_on = None
+airflow_version = "3.4.0"
+
+
+def upgrade():
+    """Apply add source_reference to import_error."""
+    with op.batch_alter_table("import_error", schema=None) as batch_op:
+        batch_op.add_column(sa.Column("source_reference", sa.String(length=2000), nullable=True))
+
+
+def downgrade():
+    """Unapply add source_reference to import_error."""
+    with op.batch_alter_table("import_error", schema=None) as batch_op:
+        batch_op.drop_column("source_reference")

airflow-core/src/airflow/migrations/versions/0140_3_4_0_add_language_to_dag_code.py

@@ -0,0 +1,52 @@
+#
+# 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 language column to dag_code.
+
+Revision ID: e5a91c7f42b3
+Revises: ca8499dc1004
+Create Date: 2026-09-10 12:00:00.000000
+
+"""
+
+from __future__ import annotations
+
+import sqlalchemy as sa
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision = "e5a91c7f42b3"
+down_revision = "ca8499dc1004"
+branch_labels = None
+depends_on = None
+airflow_version = "3.4.0"
+
+
+def upgrade():
+    """Apply add language column to dag_code."""
+    with op.batch_alter_table("dag_code", schema=None) as batch_op:
+        batch_op.add_column(
+            sa.Column("language", sa.String(length=64), nullable=False, server_default="python")
+        )
+
+
+def downgrade():
+    """Unapply add language column to dag_code."""
+    with op.batch_alter_table("dag_code", schema=None) as batch_op:
+        batch_op.drop_column("language")

airflow-core/src/airflow/models/dagcode.py

@@ -66,6 +66,9 @@ class DagCode(Base):
     )
     source_code: Mapped[str] = mapped_column(Text().with_variant(MEDIUMTEXT(), "mysql"), nullable=False)
     source_code_hash: Mapped[str] = mapped_column(String(32), nullable=False)
+    language: Mapped[str] = mapped_column(
+        String(64), nullable=False, default="python", server_default="python"
+    )
     dag_version_id: Mapped[UUID] = mapped_column(
         sa.Uuid(), ForeignKey("dag_version.id", ondelete="CASCADE"), nullable=False, unique=True
     )

airflow-core/src/airflow/models/dagwarning.py

@@ -19,8 +19,9 @@
 
 from datetime import datetime
 from enum import Enum
-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, Annotated
 
+from pydantic import StringConstraints, TypeAdapter
 from sqlalchemy import ForeignKeyConstraint, Index, String, Text, delete, select, true
 from sqlalchemy.orm import Mapped, mapped_column, relationship
 
@@ -34,6 +35,8 @@
 if TYPE_CHECKING:
     from sqlalchemy.orm import Session
 
+WARNING_TYPE_MAX_LENGTH = 50
+
 
 class DagWarning(Base):
     """
@@ -45,7 +48,7 @@ class DagWarning(Base):
     """
 
     dag_id: Mapped[str] = mapped_column(StringID(), primary_key=True)
-    warning_type: Mapped[str] = mapped_column(String(50), primary_key=True)
+    warning_type: Mapped[str] = mapped_column(String(WARNING_TYPE_MAX_LENGTH), primary_key=True)
     message: Mapped[str] = mapped_column(Text, nullable=False)
     timestamp: Mapped[datetime] = mapped_column(UtcDateTime, nullable=False, default=timezone.utcnow)
 
@@ -65,7 +68,7 @@ class DagWarning(Base):
     def __init__(self, dag_id: str, warning_type: str, message: str, **kwargs):
         super().__init__(**kwargs)
         self.dag_id = dag_id
-        self.warning_type = DagWarningType(warning_type).value  # make sure valid type
+        self.warning_type = get_warning_type_value(warning_type)
         self.message = message
 
     def __eq__(self, other) -> bool:
@@ -105,3 +108,25 @@ class DagWarningType(str, Enum):
     DUPLICATE_DAG_ID = "duplicate dag id"
     NONEXISTENT_POOL = "non-existent pool"
     RUNTIME_VARYING_VALUE = "runtime varying value"
+
+
+ImporterWarningType = Annotated[
+    str,
+    StringConstraints(pattern=r"^[a-z][a-z0-9_]*:[a-z0-9_.\-]+$", max_length=WARNING_TYPE_MAX_LENGTH),
+]
+"""Warning type reported by a Dag importer, prefixed with its namespace (e.g. ``yaml:deprecated_field``)."""
+
+DagWarningTypeValue = DagWarningType | ImporterWarningType
+"""Any valid ``warning_type``: a built-in :class:`DagWarningType` or an :data:`ImporterWarningType`."""
+
+_warning_type_adapter: TypeAdapter[DagWarningTypeValue] = TypeAdapter(DagWarningTypeValue)
+
+
+def get_warning_type_value(warning_type: str) -> str:
+    """
+    Return the value to store for ``warning_type``.
+
+    :raises ValueError: if it is neither a :class:`DagWarningType` nor an :data:`ImporterWarningType`.
+    """
+    validated = _warning_type_adapter.validate_python(warning_type)
+    return validated.value if isinstance(validated, DagWarningType) else validated