From 7152d46d4232290a21d31866981d6182696d18b3 Mon Sep 17 00:00:00 2001 From: Naveen Karanam Date: Mon, 24 Aug 2026 16:36:40 -0500 Subject: [PATCH 1/2] Add test for reparsing files with import errors --- .../routes/public/test_dag_parsing.py | 35 +++++++++++++++++++ 1 file changed, 35 insertions(+) diff --git a/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_dag_parsing.py b/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_dag_parsing.py index b60b7011a9ea3..ef04ba480b573 100644 --- a/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_dag_parsing.py +++ b/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_dag_parsing.py @@ -16,10 +16,13 @@ # under the License. from __future__ import annotations +from unittest import mock + import pytest from sqlalchemy import select from airflow.models.dagbag import DagPriorityParsingRequest, DBDagBag +from airflow.models.errors import ParseImportError from tests_common.test_utils.api_fastapi import _check_last_log from tests_common.test_utils.db import clear_db_dag_parsing_requests, clear_db_logs, parse_and_sync_to_db @@ -86,6 +89,38 @@ def test_should_respond_403(self, unauthorized_test_client, url_safe_serializer, ) assert response.status_code == 403 + @mock.patch("airflow.api_fastapi.core_api.security.get_auth_manager") + def test_reparse_import_error_file_without_registered_dag( + self, + mock_get_auth_manager, + url_safe_serializer, + session, + test_client, + ): + error = ParseImportError( + bundle_name="example_dags", + filename="broken_dag.py", + stacktrace="Broken DAG import", + ) + session.add(error) + session.commit() + + mock_get_auth_manager.return_value.authorize_view.return_value = True + + token = url_safe_serializer.dumps( + { + "bundle_name": "example_dags", + "relative_fileloc": "broken_dag.py", + } + ) + + response = test_client.put( + f"/parseDagFile/{token}", + headers={"Accept": "application/json"}, + ) + + assert response.status_code == 201 + def test_bad_file_request(self, url_safe_serializer, session, test_client): payload = {"bundle_name": "some_bundle", "relative_fileloc": "/some/random/file.py"} url = f"/parseDagFile/{url_safe_serializer.dumps(payload)}" From 20bf2c598a710ee87da78e41c4383ef23785dde6 Mon Sep 17 00:00:00 2001 From: Naveen Karanam Date: Mon, 24 Aug 2026 17:02:22 -0500 Subject: [PATCH 2/2] Add reparse support for DAG import errors --- .../api_fastapi/core_api/datamodels/dags.py | 24 ++++--- .../core_api/datamodels/import_error.py | 12 +++- .../airflow/api_fastapi/core_api/security.py | 65 ++++++++++++++++--- .../Dashboard/Stats/DagImportErrorsModal.tsx | 31 ++++++--- 4 files changed, 106 insertions(+), 26 deletions(-) diff --git a/airflow-core/src/airflow/api_fastapi/core_api/datamodels/dags.py b/airflow-core/src/airflow/api_fastapi/core_api/datamodels/dags.py index e67777d6ffa4e..887d46a951fd7 100644 --- a/airflow-core/src/airflow/api_fastapi/core_api/datamodels/dags.py +++ b/airflow-core/src/airflow/api_fastapi/core_api/datamodels/dags.py @@ -62,6 +62,13 @@ def _get_file_token_serializer() -> URLSafeSerializer: """ return URLSafeSerializer(conf.get_mandatory_value("api", "secret_key")) +def create_file_token(*, bundle_name: str | None, relative_fileloc: str | None) -> str: + """Create a signed token identifying a Dag file.""" + payload = { + "bundle_name": bundle_name, + "relative_fileloc": relative_fileloc, + } + return _get_file_token_serializer().dumps(payload) DAG_ALIAS_MAPPING: dict[str, str] = { # The keys are the names in the response, the values are the original names in the model @@ -148,15 +155,14 @@ def is_backfillable(self) -> bool: return True # Mypy issue https://github.com/python/mypy/issues/1362 - @computed_field # type: ignore[prop-decorator] - @property - def file_token(self) -> str: - """Return file token.""" - payload = { - "bundle_name": self.bundle_name, - "relative_fileloc": self.relative_fileloc, - } - return _get_file_token_serializer().dumps(payload) + @computed_field +@property +def file_token(self) -> str: + """Return file token.""" + return create_file_token( + bundle_name=self.bundle_name, + relative_fileloc=self.relative_fileloc, + ) class DAGPatchBody(StrictBaseModel): diff --git a/airflow-core/src/airflow/api_fastapi/core_api/datamodels/import_error.py b/airflow-core/src/airflow/api_fastapi/core_api/datamodels/import_error.py index 084434cadbf52..a183ca628bc61 100644 --- a/airflow-core/src/airflow/api_fastapi/core_api/datamodels/import_error.py +++ b/airflow-core/src/airflow/api_fastapi/core_api/datamodels/import_error.py @@ -19,9 +19,10 @@ from collections.abc import Iterable from datetime import datetime -from pydantic import Field +from pydantic import Field, computed_field from airflow.api_fastapi.core_api.base import BaseModel +from airflow.api_fastapi.core_api.datamodels.dags import create_file_token class ImportErrorResponse(BaseModel): @@ -33,6 +34,15 @@ class ImportErrorResponse(BaseModel): bundle_name: str | None stacktrace: str = Field(alias="stack_trace") + @computed_field + @property + def file_token(self) -> str: + """Return file token for reparsing the failed Dag file.""" + return create_file_token( + bundle_name=self.bundle_name, + relative_fileloc=self.filename, + ) + class ImportErrorCollectionResponse(BaseModel): """Import Error Collection Response.""" diff --git a/airflow-core/src/airflow/api_fastapi/core_api/security.py b/airflow-core/src/airflow/api_fastapi/core_api/security.py index 9aa1ddc783f17..7fbcd6080722e 100644 --- a/airflow-core/src/airflow/api_fastapi/core_api/security.py +++ b/airflow-core/src/airflow/api_fastapi/core_api/security.py @@ -74,7 +74,9 @@ from airflow.models.backfill import Backfill from airflow.models.dag import DagModel, DagRun, DagTag from airflow.models.dag_version import DagVersion +from airflow.models.dagbundle import DagBundleModel from airflow.models.dagwarning import DagWarning +from airflow.models.errors import ParseImportError from airflow.models.log import Log from airflow.models.taskinstance import TaskInstance as TI from airflow.models.team import Team @@ -207,7 +209,13 @@ def requires_access_dag_from_file_token( """ Authorize the caller against the DAGs referenced by a signed ``file_token``. - For ``file_token`` based endpoints (such as ``reparse``), the token is resolved to its referenced file, and authorization is performed against exactly the DAGs defined in that file, never against a request parameter. + For ``file_token`` based endpoints (such as ``reparse``), the token is + resolved to its referenced file, and authorization is performed against + exactly the DAGs defined in that file, never against a request parameter. + + If the file has an import error but no registered Dag yet, authorization + falls back to the ``IMPORT_ERRORS_ALL`` view scoped to the file's bundle + team. """ def inner( @@ -221,27 +229,68 @@ def inner( except BadSignature: raise HTTPException(status.HTTP_404_NOT_FOUND, "File not found") + bundle_name = payload["bundle_name"] + relative_fileloc = payload["relative_fileloc"] + dag_ids = list( session.scalars( select(DagModel.dag_id).where( - DagModel.bundle_name == payload["bundle_name"], - DagModel.relative_fileloc == payload["relative_fileloc"], + DagModel.bundle_name == bundle_name, + DagModel.relative_fileloc == relative_fileloc, ) ) ) + if not dag_ids: - raise HTTPException(status.HTTP_404_NOT_FOUND, "File not found") + import_error_exists = session.scalar( + select(ParseImportError.id).where( + ParseImportError.bundle_name == bundle_name, + ParseImportError.filename == relative_fileloc, + ) + ) + + if import_error_exists is None: + raise HTTPException(status.HTTP_404_NOT_FOUND, "File not found") + + team_name = ( + DagBundleModel.get_team_name(bundle_name, session=session) + if bundle_name + else None + ) + + _requires_access( + is_authorized_callback=lambda: get_auth_manager().authorize_view( + access_view=AccessView.IMPORT_ERRORS_ALL, + user=user, + team_name=team_name, + ), + ) + return + + dag_id_to_team = DagModel.get_dag_id_to_team_name_mapping( + dag_ids, + session=session, + ) - dag_id_to_team = DagModel.get_dag_id_to_team_name_mapping(dag_ids, session=session) requests: list[IsAuthorizedDagRequest] = [ - {"method": method, "details": DagDetails(id=dag_id, team_name=dag_id_to_team.get(dag_id))} + { + "method": method, + "details": DagDetails( + id=dag_id, + team_name=dag_id_to_team.get(dag_id), + ), + } for dag_id in dag_ids ] + _requires_access( - is_authorized_callback=lambda: get_auth_manager().batch_is_authorized_dag(requests, user=user), + is_authorized_callback=lambda: get_auth_manager().batch_is_authorized_dag( + requests, + user=user, + ), ) - return inner + return inner class PermittedDagFilter(OrmClause[set[str]]): diff --git a/airflow-core/src/airflow/ui/src/pages/Dashboard/Stats/DagImportErrorsModal.tsx b/airflow-core/src/airflow/ui/src/pages/Dashboard/Stats/DagImportErrorsModal.tsx index 6a1a142966913..453b67e881f46 100644 --- a/airflow-core/src/airflow/ui/src/pages/Dashboard/Stats/DagImportErrorsModal.tsx +++ b/airflow-core/src/airflow/ui/src/pages/Dashboard/Stats/DagImportErrorsModal.tsx @@ -22,10 +22,16 @@ import { useTranslation } from "react-i18next"; import { LuFileWarning } from "react-icons/lu"; import { PiFilePy } from "react-icons/pi"; -import { useImportErrorServiceGetImportErrors } from "openapi/queries"; +import { + useDagParsingServiceReparseDagFile, + useImportErrorServiceGetImportErrors, +} from "openapi/queries"; +import { AiOutlineFileSync } from "react-icons/ai"; +import { IconButton } from "src/components/ui"; import { SearchBar } from "src/components/SearchBar"; import Time from "src/components/Time"; -import { Accordion, ClipboardIconButton, Modal } from "src/components/ui"; +import { Accordion, ClipboardIconButton, IconButton, Modal } from "src/components/ui"; +import { AiOutlineFileSync } from "react-icons/ai"; import { Pagination } from "src/components/ui/Pagination"; type ImportDAGErrorModalProps = { @@ -50,7 +56,7 @@ export const DagImportErrorsModal = ({ onClose, open }: ImportDAGErrorModalProps ); const { t: translate } = useTranslation(["dashboard", "components"]); - + const { isPending, mutate } = useDagParsingServiceReparseDagFile(); const onOpenChange = () => { setSearchQuery(""); setPage(1); @@ -119,11 +125,20 @@ export const DagImportErrorsModal = ({ onClose, open }: ImportDAGErrorModalProps {importError.filename} - - - - - + + mutate({ fileToken: importError.file_token })} + variant="outline" + > + + + + + + +