Add reparse support for DAG import errors

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):

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."""

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]]):

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}
                 </HStack>
               </Accordion.ItemTrigger>
-              <Box alignItems="center" display="flex" flexShrink={0} pr={2}>
-                <ClipboardRoot value={importError.filename}>
-                  <ClipboardIconButton variant="outline" />
-                </ClipboardRoot>
-              </Box>
+              <Box alignItems="center" display="flex" flexShrink={0} gap={2} pr={2}>
+  <IconButton
+    label={translate("components:reparseDag")}
+    loading={isPending}
+    onClick={() => mutate({ fileToken: importError.file_token })}
+    variant="outline"
+  >
+    <AiOutlineFileSync />
+  </IconButton>
+
+  <ClipboardRoot value={importError.filename}>
+    <ClipboardIconButton variant="outline" />
+  </ClipboardRoot>
+</Box>
             </HStack>
             <Accordion.ItemContent>
               <Text color="fg.muted" fontSize="sm" mb={1}>

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)}"