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