shahar1 commented on code in PR #75673:
URL: https://github.com/apache/airflow/pull/75673#discussion_r4433215249
##########
airflow-core/src/airflow/api_fastapi/core_api/security.py:
##########
@@ -315,8 +315,24 @@ def to_orm(self, select: Select) -> Select:
class PermittedDagWarningFilter(PermittedDagFilter):
"""A parameter that filters the permitted dag warnings for the user."""
- def to_orm(self, select: Select) -> Select:
- return select.where(DagWarning.dag_id.in_(self.value or set()))
+ def to_orm(self, statement: Select) -> Select:
+ authorized = self.value or set()
+ readable_files = (
+ select(DagModel.bundle_name, DagModel.relative_fileloc)
+ .where(DagModel.dag_id.in_(authorized))
+ .distinct()
+ )
+ return statement.where(
+ or_(
+ DagWarning.dag_id.in_(authorized),
Review Comment:
Anyone who can read one Dag in the file sees the stacktrace. For import
errors from the same file, the API hides
the stacktrace unless the user can read all Dags in it. If no Dag in the
file loaded, nobody can see the warning.
##########
airflow-core/src/airflow/dag_processing/collection.py:
##########
@@ -383,6 +387,58 @@ def _update_dag_warnings(
session.merge(warning_to_add)
+def dag_parse_failure_warnings(bundle_name: str, warnings_by_rel_path:
dict[str, list]) -> list[DagWarning]:
+ return [
+ DagWarning(
+ dag_id=record.dag_id,
+ warning_type=record.warning_type,
+ message=record.message,
+ bundle_name=bundle_name,
+ filename=rel_path,
+ )
+ for rel_path, records in warnings_by_rel_path.items()
+ for record in records
+ ]
+
+
+def _update_file_scoped_parse_failure_warnings(
+ files_parsed: set[tuple[str, str]],
+ parse_failure_warnings: Iterable[DagWarning],
+ session: Session,
+) -> None:
+ if files_parsed:
+ session.execute(
+ delete(DagWarning).where(
+ tuple_(DagWarning.bundle_name,
DagWarning.filename).in_(files_parsed),
+ DagWarning.warning_type == PARSE_FAILURE_WARNING_TYPE,
Review Comment:
1. `emit_dag_warning()` accepts a custom `warning_type, but the cleanup
functions, the
access filter and the Dashboard card only look at
`PARSE_FAILURE_WARNING_TYPE`.
2. The cleanup for inactive Dags also skips
these rows, so a warning with a custom type stays in the table forever.
##########
task-sdk/src/airflow/sdk/importers/python_importer.py:
##########
@@ -211,6 +212,9 @@ def import_definition(
)
self._process_modules(modules, result, bundle=bundle)
+ from airflow.sdk.definitions.parse_warning import drain_parse_warnings
Review Comment:
Could you extract the import to the top-level of this module?
##########
airflow-core/src/airflow/dag_processing/collection.py:
##########
@@ -383,6 +387,58 @@ def _update_dag_warnings(
session.merge(warning_to_add)
+def dag_parse_failure_warnings(bundle_name: str, warnings_by_rel_path:
dict[str, list]) -> list[DagWarning]:
Review Comment:
Could you please rename the function so it would start with a verb?
##########
airflow-core/src/airflow/models/dagwarning.py:
##########
@@ -51,25 +52,35 @@ class DagWarning(Base):
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)
-
- dag_model = relationship("DagModel", viewonly=True, lazy="selectin")
-
- __tablename__ = "dag_warning"
- __table_args__ = (
- ForeignKeyConstraint(
- ("dag_id",),
- ["dag.dag_id"],
- name="dcw_dag_id_fkey",
- ondelete="CASCADE",
- ),
- Index("idx_dag_warning_dag_id", dag_id),
+ bundle_name: Mapped[str | None] = mapped_column(StringID(), nullable=True)
+ filename: Mapped[str | None] = mapped_column(String(1024), nullable=True)
Review Comment:
Every parse now runs an extra DELETE on `(bundle_name, filename)`, and
neither column is indexed. The PR raised the
statement counts in `test_manager.py` to fit it.
##########
task-sdk/src/airflow/sdk/execution_time/schema/versions/v2026_10_30.py:
##########
@@ -105,6 +105,13 @@ class
AddDagDefinitionsToDagFileParsingResult(VersionChange):
)
+class AddParseWarningsToDagFileParsingResult(VersionChange):
Review Comment:
Could you please add a docstring? (otherwise `description` below is `None`)
##########
airflow-core/src/airflow/ui/src/pages/Dashboard/Stats/ParseFailureWarnings.tsx:
##########
@@ -0,0 +1,73 @@
+/*!
+ * 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.
+ */
+import { Skeleton, useDisclosure } from "@chakra-ui/react";
+import { useTranslation } from "react-i18next";
+import { LuFileWarning } from "react-icons/lu";
+
+import { useDagWarningServiceListDagWarnings } from "openapi/queries";
+
+import { DAGWarningsModal } from "src/components/DAGWarningsModal";
+import type { ExpandedApiError } from "src/components/ErrorAlert";
+import { StatsCard } from "src/components/StatsCard";
+
+import { useAutoRefresh } from "src/utils";
+
+const PARSE_FAILURE_WARNING_TYPE = "user:dynamic_dag_failure";
Review Comment:
You repeat the literal `user:dynamic_dag_failure` few times in the code
rather importing it from a single source of truth, please consider changing
that where it's not necessary to repeat.
##########
airflow-core/src/airflow/dag_processing/collection.py:
##########
@@ -383,6 +387,58 @@ def _update_dag_warnings(
session.merge(warning_to_add)
+def dag_parse_failure_warnings(bundle_name: str, warnings_by_rel_path:
dict[str, list]) -> list[DagWarning]:
+ return [
+ DagWarning(
+ dag_id=record.dag_id,
+ warning_type=record.warning_type,
+ message=record.message,
+ bundle_name=bundle_name,
+ filename=rel_path,
+ )
+ for rel_path, records in warnings_by_rel_path.items()
+ for record in records
+ ]
+
+
+def _update_file_scoped_parse_failure_warnings(
+ files_parsed: set[tuple[str, str]],
+ parse_failure_warnings: Iterable[DagWarning],
+ session: Session,
+) -> None:
Review Comment:
nit: `session` should be keyword-only argument
```suggestion
def _update_file_scoped_parse_failure_warnings(
*,
files_parsed: set[tuple[str, str]],
parse_failure_warnings: Iterable[DagWarning],
session: Session,
) -> None:
```
##########
task-sdk/src/airflow/sdk/definitions/parse_warning.py:
##########
@@ -0,0 +1,55 @@
+# 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.
+from __future__ import annotations
+
+import re
+
+from airflow.sdk.definitions.parse_warning_record import DagParseWarningRecord
+
+_PARSE_WARNING_TYPE_RE = re.compile(r"^[a-z][a-z0-9_]*:[a-z0-9_.\-]+$",
re.ASCII)
+_DEFAULT_WARNING_TYPE = "user:dynamic_dag_failure"
+
+
+def emit_dag_warning(
Review Comment:
There are no tests for task-sdk at all - this one needs to be well-covered,
since it's the essence of this PR.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
