kaxil commented on code in PR #69864:
URL: https://github.com/apache/airflow/pull/69864#discussion_r3962879415
##########
airflow-core/src/airflow/models/dag_version.py:
##########
@@ -245,6 +255,179 @@ def version(self) -> str:
"""A human-friendly representation of the version."""
return f"{self.dag_id}-{self.version_number}"
+ @classmethod
+ @provide_session
+ def get_diff(
+ cls,
+ dag_id: str,
+ base_version_number: int,
+ target_version_number: int,
+ *,
+ include_values: bool = False,
+ include_source: bool = False,
+ max_changes: int | None = None,
+ source_status: SourceStatus | None = None,
+ values_status: ValuesStatus | None = None,
+ session: Session = NEW_SESSION,
+ ) -> dict[str, Any]:
+ """
+ Compare two versions of a Dag using their currently stored state.
+
+ ``source_status`` is supplied by callers that have an authorization
context. The CLI has
+ operator-level authority and leaves it unset, while API callers
calculate it before this
+ method returns any source content. API callers use ``values_status``
to suppress raw values
+ when the user cannot access Dag code; the structural diff remains
available in redacted form.
+ """
+ # Keep this local to avoid the dag_version -> dag_version_diff ->
serialized_objects cycle.
+ from airflow.models.serialized_dag import SerializedDagModel
+ from airflow.serialization.dag_version_diff import (
+ DEFAULT_MAX_CHANGES,
+ build_serialized_dag_diff,
+ )
+
+ if source_status is not None and source_status not in
_VALID_SOURCE_STATUSES:
+ raise ValueError(f"source_status must be one of
{_VALID_SOURCE_STATUSES}, not {source_status!r}")
+ if values_status is not None and values_status not in
_VALID_VALUES_STATUSES:
+ raise ValueError(f"values_status must be one of
{_VALID_VALUES_STATUSES}, not {values_status!r}")
+
+ if max_changes is None:
+ max_changes = DEFAULT_MAX_CHANGES
+ if base_version_number < 1 or target_version_number < 1:
+ raise ValueError("Dag version numbers must be positive integers")
+
+ if source_status is None:
+ source_status = "current_stored_code" if include_source else
"unavailable"
+
+ query = (
+ select(cls)
+ .where(
+ cls.dag_id == dag_id,
+ cls.version_number.in_((base_version_number,
target_version_number)),
+ )
+
.options(joinedload(cls.serialized_dag).selectinload(SerializedDagModel.deadline_alerts))
+ )
+ if include_source and source_status == "current_stored_code":
+ query = query.options(joinedload(cls.dag_code))
+
+ versions = {version.version_number: version for version in
session.scalars(query).all()}
+ missing_version = next(
+ (
+ version_number
+ for version_number in (base_version_number,
target_version_number)
+ if version_number not in versions
+ ),
+ None,
+ )
+ if missing_version is not None:
+ raise DagVersionNotFound(
+ f"The DagVersion with dag_id: `{dag_id}` and version_number:
`{missing_version}` was not found"
+ )
+
+ base_version = versions[base_version_number]
+ target_version = versions[target_version_number]
+ base_data, base_unavailable_reason =
_get_serialized_diff_data(base_version.serialized_dag)
+ target_data, target_unavailable_reason =
_get_serialized_diff_data(target_version.serialized_dag)
+ effective_include_values = include_values and values_status in {None,
"available"}
+ result = build_serialized_dag_diff(
+ base_data=base_data,
+ target_data=target_data,
+ base_provenance=_get_provenance(base_version),
+ target_provenance=_get_provenance(target_version),
+ include_values=effective_include_values,
+ max_changes=max_changes,
+ )
+ if unavailable_reason := base_unavailable_reason or
target_unavailable_reason:
+ result.update(
+ mode="unavailable", unavailable_reason=unavailable_reason,
changes=[], truncated=False
+ )
+ if include_values:
+ values_available = effective_include_values and result["mode"] ==
"observed_state"
+ result["values"] = {"status": "available" if values_available else
"unavailable"}
Review Comment:
`values` only shows up when `include_values` is set, while `source` is
always present with `status: "unavailable"` when it's off. That means
`result["source"]["status"]` is always safe to read but
`result["values"]["status"]` raises KeyError on a redacted request. Emitting
`values` unconditionally would keep the `diff_schema_version: 1` shape stable
for clients.
##########
airflow-core/src/airflow/serialization/dag_version_diff.py:
##########
@@ -0,0 +1,655 @@
+# 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.
+
+"""Observed-state diffs for serialized Dag payloads."""
+
+from __future__ import annotations
+
+import copy
+import hashlib
+import json
+from collections.abc import Callable, Mapping
+from enum import Enum
+from typing import Any, Literal
+
+from airflow.serialization.definitions.baseoperator import
SerializedBaseOperator
+from airflow.serialization.definitions.mappedoperator import
SerializedMappedOperator
+from airflow.serialization.serialized_objects import (
+ _DAG_CALLBACK_FIELDS,
+ _OPERATOR_TIMEDELTA_FIELDS,
+ DagSerialization,
+ OperatorSerialization,
+)
+
+DIFF_SCHEMA_VERSION = 1
+DEFAULT_MAX_CHANGES = 500
+MAX_ALLOWED_CHANGES = 5000
+SUPPORTED_SERIALIZED_DAG_SCHEMA_VERSIONS = frozenset((1, 2, 3))
+
+DiffMode = Literal["observed_state", "unavailable"]
+
+_ORDER_INSENSITIVE_LIST_PATHS = {
+ ("dag", "tags"),
+ ("dag", "allowed_run_types"),
+}
+_KEYED_COLLECTION_PATHS = {
+ ("dag", "tasks"),
+ ("dag", "dag_dependencies"),
+ *_ORDER_INSENSITIVE_LIST_PATHS,
+}
+_CUSTOM_TASK_FIELDS_PATH_COMPONENT = "custom_fields"
+# This allowlist is part of diff schema v1. Serializer schema changes must not
+# silently change the paths visible to callers of the diff API.
+_DIFF_V1_PUBLIC_TASK_FIELDS = frozenset(
+ {
+ "__type",
+ "_disallow_kwargs_override",
+ "_expand_input_attr",
+ "_is_mapped",
+ "_is_sensor",
+ "_logger_name",
+ "_needs_expansion",
+ "_operator_extra_links",
+ "_task_display_name",
+ "_task_module",
+ "allow_nested_operators",
+ "depends_on_past",
+ "do_xcom_push",
+ "doc",
+ "doc_json",
+ "doc_md",
+ "doc_rst",
+ "doc_yaml",
+ "downstream_task_ids",
+ "email_on_failure",
+ "email_on_retry",
+ "end_date",
+ "execution_timeout",
+ "executor",
+ "executor_config",
+ "has_on_execute_callback",
+ "has_on_failure_callback",
+ "has_on_retry_callback",
+ "has_on_skipped_callback",
+ "has_on_success_callback",
+ "ignore_first_depends_on_past",
+ "inlets",
+ "is_setup",
+ "is_teardown",
+ "map_index_template",
+ "max_active_tis_per_dag",
+ "max_active_tis_per_dagrun",
+ "max_retry_delay",
+ "multiple_outputs",
+ "on_failure_fail_dagrun",
+ "outlets",
+ "owner",
+ "params",
+ "partial_kwargs",
+ "pool",
+ "pool_slots",
+ "priority_weight",
+ "queue",
+ "render_template_as_native_obj",
+ "retries",
+ "retry_delay",
+ "retry_exponential_backoff",
+ "start_date",
+ "start_from_trigger",
+ "start_trigger_args",
+ "task_id",
+ "task_type",
+ "template_ext",
+ "template_fields",
+ "template_fields_renderers",
+ "trigger_rule",
+ "ui_color",
+ "ui_fgcolor",
+ "wait_for_downstream",
+ "wait_for_past_depends_before_skipping",
+ "weight_rule",
+ }
+)
+_DIFF_V1_REDACTED_SCHEMA_TASK_FIELDS = frozenset({"_arg_bindings"})
+_DIFF_V1_PUBLIC_PARTIAL_TASK_FIELDS = _DIFF_V1_PUBLIC_TASK_FIELDS |
{"task_display_name"}
+# Classify every Dag schema field explicitly so new fields require a policy
decision.
+_DIFF_V1_DAG_FIELD_CATEGORIES = {
+ "_concurrency": "schedule",
+ "_processor_dags_folder": "provenance",
+ "access_control": "authorization",
+ "allowed_run_types": "schedule",
+ "bundle_name": "provenance",
+ "catchup": "schedule",
+ "dag_dependencies": "dependency",
+ "dag_display_name": "metadata",
+ "dag_id": "metadata",
+ "dagrun_timeout": "schedule",
+ "deadline": "deadline",
+ "default_args": "param",
+ "description": "metadata",
+ "disable_bundle_versioning": "task",
+ "doc_md": "metadata",
+ "edge_info": "metadata",
+ "end_date": "schedule",
+ "fail_fast": "schedule",
+ "fileloc": "provenance",
+ "has_on_failure_callback": "callback",
+ "has_on_success_callback": "callback",
+ "is_paused_upon_creation": "schedule",
+ "max_active_runs": "schedule",
+ "max_active_tasks": "schedule",
+ "max_consecutive_failed_dag_runs": "schedule",
+ "owner_links": "metadata",
+ "params": "param",
+ "relative_fileloc": "provenance",
+ "render_template_as_native_obj": "task",
+ "rerun_with_latest_version": "task",
+ "start_date": "schedule",
+ "tags": "metadata",
+ "task_group": "task",
+ "tasks": "task",
+ "timetable": "schedule",
+ "timezone": "schedule",
+}
+_DIFF_V1_LEGACY_DAG_FIELD_CATEGORIES = {
+ "fail_stop": "schedule",
+ "on_failure_callback": "callback",
+ "on_success_callback": "callback",
+ "schedule": "schedule",
+ "schedule_interval": "schedule",
+}
+_REDACTED_RECURSIVE_MAPPING_PATHS = {
+ (),
+ ("dag",),
+ ("dag", "task_group"),
+ ("provenance",),
+ *_KEYED_COLLECTION_PATHS,
+}
+
+
+def build_serialized_dag_diff(
+ *,
+ base_data: dict[str, Any] | None,
+ target_data: dict[str, Any] | None,
+ base_provenance: Mapping[str, Any] | None = None,
+ target_provenance: Mapping[str, Any] | None = None,
+ include_values: bool = False,
+ max_changes: int = DEFAULT_MAX_CHANGES,
+) -> dict[str, Any]:
+ """
+ Build a bounded, deterministic diff from two stored serialized Dag
payloads.
+
+ Raw values, digests, and value-derived path components are returned only
when
+ ``include_values`` is true.
+ """
+ _validate_max_changes(max_changes)
+
+ base_schema_version = _get_schema_version(base_data)
+ target_schema_version = _get_schema_version(target_data)
+ result: dict[str, Any] = {
+ "diff_schema_version": DIFF_SCHEMA_VERSION,
+ "serialized_dag_schema_versions": {
+ "base": base_schema_version,
+ "target": target_schema_version,
+ },
+ "mode": "observed_state",
+ "changes": [],
+ "truncated": False,
+ }
+
+ if base_data is None or target_data is None:
+ return _mark_unavailable(result, "serialized_dag_missing")
+
+ if base_schema_version is None or target_schema_version is None:
+ return _mark_unavailable(result,
"serialized_dag_schema_version_missing")
+
+ unsupported_versions = [
+ version
+ for version in (base_schema_version, target_schema_version)
+ if version not in SUPPORTED_SERIALIZED_DAG_SCHEMA_VERSIONS
+ ]
+ if unsupported_versions:
+ return _mark_unavailable(
+ result,
f"unsupported_serialized_dag_schema_version:{unsupported_versions[0]}"
+ )
+
+ try:
+ base_document = _canonicalize_payload_v1(base_data)
+ target_document = _canonicalize_payload_v1(target_data)
+ except (AttributeError, KeyError, OverflowError, TypeError, ValueError):
+ return _mark_unavailable(result,
"serialized_dag_canonicalization_failed")
+
+ base_document["provenance"] = _canonicalize_value(dict(base_provenance or
{}), path=("provenance",))
+ target_document["provenance"] = _canonicalize_value(dict(target_provenance
or {}), path=("provenance",))
+
+ collector = _ChangeCollector(max_changes=max_changes,
include_values=include_values)
+ _collect_changes(base_document, target_document, path=(),
collector=collector)
+ result["changes"] = collector.changes
+ result["truncated"] = collector.is_truncated
+ return result
+
+
+class _ChangeCollector:
+ def __init__(self, *, max_changes: int, include_values: bool) -> None:
+ self.changes: list[dict[str, Any]] = []
+ self.count = 0
+ self.max_changes = max_changes
+ self.include_values = include_values
+
+ @property
+ def is_truncated(self) -> bool:
+ return self.count > self.max_changes
+
+ def add(
+ self,
+ *,
+ path: tuple[str, ...],
+ operation: Literal["added", "removed", "changed"],
+ before: Any,
+ after: Any,
+ ) -> None:
+ self.count += 1
+ if len(self.changes) >= self.max_changes:
+ return
+
+ public_path = _get_public_path(path)
+ category = _get_category(public_path)
+ change: dict[str, Any] = {
+ "path": _format_path(path if self.include_values else public_path),
+ "operation": operation,
+ "category": category,
+ "impact": _get_impact(category),
+ }
+ if self.include_values:
+ change["before_digest"] = None if before is _MISSING else
_get_digest(before)
+ change["after_digest"] = None if after is _MISSING else
_get_digest(after)
+ if before is not _MISSING:
+ change["before_value"] = before
+ if after is not _MISSING:
+ change["after_value"] = after
+ self.changes.append(change)
+
+
+_MISSING = object()
+
+
+def _validate_max_changes(max_changes: int) -> None:
+ if max_changes < 1:
+ raise ValueError("max_changes must be a positive integer")
+ if max_changes > MAX_ALLOWED_CHANGES:
+ raise ValueError(f"max_changes must not exceed {MAX_ALLOWED_CHANGES}")
+
+
+def _get_schema_version(data: Mapping[str, Any] | None) -> int | None:
+ if not isinstance(data, Mapping):
+ return None
+ version = data.get("__version")
+ return version if isinstance(version, int) and not isinstance(version,
bool) else None
+
+
+def _mark_unavailable(result: dict[str, Any], reason: str) -> dict[str, Any]:
+ result["mode"] = "unavailable"
+ result["unavailable_reason"] = reason
+ return result
+
+
+def _canonicalize_payload_v1(data: dict[str, Any]) -> dict[str, Any]:
+ payload = copy.deepcopy(data)
+ version = _get_schema_version(payload)
+ if version is None:
+ raise ValueError("missing or invalid __version")
+ if version == 1:
+ DagSerialization.conversion_v1_to_v2(payload)
+ DagSerialization.conversion_v2_to_v3(payload)
+ elif version == 2:
+ DagSerialization.conversion_v2_to_v3(payload)
+ if not isinstance(payload.get("dag"), Mapping):
+ raise ValueError("missing dag object")
+ dag_defaults = {
+ field: value
+ for field, value in DagSerialization.get_schema_defaults("dag").items()
+ # Dag callback flags are enabled by their presence, even when their
value is false.
+ if field not in _DAG_CALLBACK_FIELDS
+ }
+ payload["dag"] = {**dag_defaults, **payload["dag"]}
+ _apply_task_defaults(payload)
+ payload.pop("__version", None)
+ return _canonicalize_value(payload, path=())
+
+
+def _apply_task_defaults(payload: dict[str, Any]) -> None:
+ client_defaults = payload.pop("client_defaults", None)
+ if client_defaults is None:
+ client_defaults = {}
+ if not isinstance(client_defaults, Mapping):
+ raise ValueError("client_defaults is not an object")
+
+ task_defaults = client_defaults.get("tasks", {})
+ if not isinstance(task_defaults, Mapping):
+ raise ValueError("client_defaults.tasks is not an object")
+
+ schema_defaults = DagSerialization.get_schema_defaults("operator")
+ partial_fields = (
+ SerializedBaseOperator.get_serialized_fields() -
SerializedMappedOperator.get_serialized_fields()
+ )
+ tasks = payload["dag"].get("tasks", [])
+ if not isinstance(tasks, list):
+ raise ValueError("dag.tasks is not a list")
+ for task in tasks:
+ if not isinstance(task, dict) or not isinstance(task.get("__var"),
Mapping):
+ raise ValueError("task entry is not an object")
+ encoded_task = OperatorSerialization._apply_defaults_to_encoded_op(
+ dict(task["__var"]), dict(client_defaults)
+ )
+ task_data = {**schema_defaults,
**OperatorSerialization._upgrade_encoded_operator(encoded_task)}
+ if task_data.get("_is_mapped"):
+ partial_kwargs = task_data.get("partial_kwargs", {})
+ if not isinstance(partial_kwargs, Mapping):
+ raise ValueError("partial_kwargs is not an object")
+ effective_partial_kwargs = dict(task_defaults)
Review Comment:
I think the precedence is inverted here for mapped tasks. `task_data`
already has `client_defaults.tasks` merged underneath it by
`_apply_defaults_to_encoded_op` above, so seeding from `task_defaults` puts the
client default in first, and the `setdefault` on line 367 can then never
replace it with the explicit top-level value. `populate_operator` resolves the
other way: `partial_kwargs` wins, then the top-level value, then the client
default.
The effect is that real changes go missing rather than showing up wrong.
With `client_defaults.tasks = {"retries": 5}` and two mapped-task payloads
differing only in top-level `retries: 2` vs `retries: 3`,
`build_serialized_dag_diff` returns an empty `changes` list, while
`populate_operator` resolves 2 and 3. `pool` behaves the same way.
`effective_partial_kwargs = dict(partial_kwargs)` (dropping the
`task_defaults` seed and the `update`) matched `populate_operator` on every
case I tried, and the whole `test_dag_version_diff.py` suite still passes with
it, which suggests a gap: the mapped-default tests only shadow through
`partial_kwargs`, never through a top-level field.
##########
airflow-core/src/airflow/models/dag_version.py:
##########
@@ -245,6 +255,179 @@ def version(self) -> str:
"""A human-friendly representation of the version."""
return f"{self.dag_id}-{self.version_number}"
+ @classmethod
+ @provide_session
+ def get_diff(
+ cls,
+ dag_id: str,
+ base_version_number: int,
+ target_version_number: int,
+ *,
+ include_values: bool = False,
+ include_source: bool = False,
+ max_changes: int | None = None,
+ source_status: SourceStatus | None = None,
+ values_status: ValuesStatus | None = None,
+ session: Session = NEW_SESSION,
+ ) -> dict[str, Any]:
+ """
+ Compare two versions of a Dag using their currently stored state.
+
+ ``source_status`` is supplied by callers that have an authorization
context. The CLI has
+ operator-level authority and leaves it unset, while API callers
calculate it before this
+ method returns any source content. API callers use ``values_status``
to suppress raw values
+ when the user cannot access Dag code; the structural diff remains
available in redacted form.
+ """
+ # Keep this local to avoid the dag_version -> dag_version_diff ->
serialized_objects cycle.
+ from airflow.models.serialized_dag import SerializedDagModel
+ from airflow.serialization.dag_version_diff import (
+ DEFAULT_MAX_CHANGES,
+ build_serialized_dag_diff,
+ )
+
+ if source_status is not None and source_status not in
_VALID_SOURCE_STATUSES:
+ raise ValueError(f"source_status must be one of
{_VALID_SOURCE_STATUSES}, not {source_status!r}")
+ if values_status is not None and values_status not in
_VALID_VALUES_STATUSES:
+ raise ValueError(f"values_status must be one of
{_VALID_VALUES_STATUSES}, not {values_status!r}")
+
+ if max_changes is None:
+ max_changes = DEFAULT_MAX_CHANGES
Review Comment:
`max_changes` isn't range-checked until `build_serialized_dag_diff` runs, so
an out-of-range value pays for the version lookup and both serialized payload
loads before raising. Calling `_validate_max_changes` here instead would reject
it before any DB work, which will matter more once `max_changes` comes in as a
query param.
##########
airflow-core/src/airflow/models/dag_version.py:
##########
@@ -245,6 +255,179 @@ def version(self) -> str:
"""A human-friendly representation of the version."""
return f"{self.dag_id}-{self.version_number}"
+ @classmethod
+ @provide_session
+ def get_diff(
+ cls,
+ dag_id: str,
+ base_version_number: int,
+ target_version_number: int,
+ *,
+ include_values: bool = False,
+ include_source: bool = False,
+ max_changes: int | None = None,
+ source_status: SourceStatus | None = None,
+ values_status: ValuesStatus | None = None,
+ session: Session = NEW_SESSION,
+ ) -> dict[str, Any]:
+ """
+ Compare two versions of a Dag using their currently stored state.
+
+ ``source_status`` is supplied by callers that have an authorization
context. The CLI has
+ operator-level authority and leaves it unset, while API callers
calculate it before this
+ method returns any source content. API callers use ``values_status``
to suppress raw values
+ when the user cannot access Dag code; the structural diff remains
available in redacted form.
+ """
+ # Keep this local to avoid the dag_version -> dag_version_diff ->
serialized_objects cycle.
+ from airflow.models.serialized_dag import SerializedDagModel
+ from airflow.serialization.dag_version_diff import (
+ DEFAULT_MAX_CHANGES,
+ build_serialized_dag_diff,
+ )
+
+ if source_status is not None and source_status not in
_VALID_SOURCE_STATUSES:
+ raise ValueError(f"source_status must be one of
{_VALID_SOURCE_STATUSES}, not {source_status!r}")
+ if values_status is not None and values_status not in
_VALID_VALUES_STATUSES:
+ raise ValueError(f"values_status must be one of
{_VALID_VALUES_STATUSES}, not {values_status!r}")
+
+ if max_changes is None:
+ max_changes = DEFAULT_MAX_CHANGES
+ if base_version_number < 1 or target_version_number < 1:
+ raise ValueError("Dag version numbers must be positive integers")
+
+ if source_status is None:
+ source_status = "current_stored_code" if include_source else
"unavailable"
Review Comment:
This defaults to full disclosure when a caller doesn't pass `source_status`,
and `values_status=None` on line 330 does the same for raw values. An API
handler that wires `include_source` to a query param but forgets to compute
`source_status` would serve DAG source to whoever can reach the endpoint, and
nothing in here would catch it.
Given the REST endpoint is the next PR in the stack, is it worth making
`source_status` required and having the CLI pass `"current_stored_code"`
explicitly? Omitting it becomes a TypeError rather than a disclosure. This is
separate from the validation point in #discussion_r3580125653, which was about
mistyped status values rather than the default direction.
--
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]
