Copilot commented on code in PR #70446:
URL: https://github.com/apache/airflow/pull/70446#discussion_r3650615499
##########
airflow-core/src/airflow/cli/commands/dag_command.py:
##########
@@ -79,6 +81,89 @@
# Chunk size for bulk delete.
_RUN_CHUNK_SIZE = 500
+def _normalize_serialized_dag_for_stability_check(serialized_dag: dict[str,
Any]) -> dict[str, Any]:
Review Comment:
PEP 8 / ruff will flag this block because top-level function definitions
must be preceded by two blank lines. Add an extra blank line between the
module-level constant and the first helper.
##########
airflow-core/src/airflow/cli/commands/dag_command.py:
##########
@@ -79,6 +81,89 @@
# Chunk size for bulk delete.
_RUN_CHUNK_SIZE = 500
+def _normalize_serialized_dag_for_stability_check(serialized_dag: dict[str,
Any]) -> dict[str, Any]:
+ normalized = SerializedDagModel._sort_serialized_dag_dict(serialized_dag)
+ normalized["dag"].pop("fileloc", None)
+ normalized["dag"].pop("bundle_name", None)
+ return normalized
+
+def _serialize_dag_for_stability_check(dag: DAG) -> tuple[str, dict[str, Any]]:
+ serialized_dag = DagSerialization.to_dict(dag)
+ return SerializedDagModel.hash(serialized_dag),
_normalize_serialized_dag_for_stability_check(serialized_dag)
+
+def _format_stability_diff(
+ dag_id: str,
+ first_serialized_dag: dict[str, Any],
+ second_serialized_dag: dict[str, Any],
+) -> str:
+ before = json.dumps(first_serialized_dag, indent=2,
sort_keys=True).splitlines()
+ after = json.dumps(second_serialized_dag, indent=2,
sort_keys=True).splitlines()
+ diff = difflib.unified_diff(
+ before,
+ after,
+ fromfile=f"parse 1: {dag_id} ",
+ tofile=f"parse 2: {dag_id} ",
+ lineterm="",
+ )
+ return "\n".join(diff)
+
+def _parse_dags_for_stability_check(dag_folder: str | None) -> DagBag:
+ return DagBag(dag_folder=dag_folder, safe_mode=True, load_op_links=False)
Review Comment:
The stability check currently forces DagBag safe_mode=True, which overrides
the configured DAG_DISCOVERY_SAFE_MODE and can silently skip Dag files. This
makes the CLI check incomplete in deployments that disable safe_mode (or want
it honored).
##########
airflow-core/src/airflow/cli/commands/dag_command.py:
##########
@@ -79,6 +81,89 @@
# Chunk size for bulk delete.
_RUN_CHUNK_SIZE = 500
+def _normalize_serialized_dag_for_stability_check(serialized_dag: dict[str,
Any]) -> dict[str, Any]:
+ normalized = SerializedDagModel._sort_serialized_dag_dict(serialized_dag)
+ normalized["dag"].pop("fileloc", None)
+ normalized["dag"].pop("bundle_name", None)
+ return normalized
+
+def _serialize_dag_for_stability_check(dag: DAG) -> tuple[str, dict[str, Any]]:
+ serialized_dag = DagSerialization.to_dict(dag)
+ return SerializedDagModel.hash(serialized_dag),
_normalize_serialized_dag_for_stability_check(serialized_dag)
+
+def _format_stability_diff(
+ dag_id: str,
+ first_serialized_dag: dict[str, Any],
+ second_serialized_dag: dict[str, Any],
+) -> str:
+ before = json.dumps(first_serialized_dag, indent=2,
sort_keys=True).splitlines()
+ after = json.dumps(second_serialized_dag, indent=2,
sort_keys=True).splitlines()
+ diff = difflib.unified_diff(
+ before,
+ after,
+ fromfile=f"parse 1: {dag_id} ",
+ tofile=f"parse 2: {dag_id} ",
+ lineterm="",
+ )
+ return "\n".join(diff)
+
+def _parse_dags_for_stability_check(dag_folder: str | None) -> DagBag:
+ return DagBag(dag_folder=dag_folder, safe_mode=True, load_op_links=False)
+
+@cli_utils.action_cli
+@providers_configuration_loaded
+def dag_stability_check(args) -> None:
+ dag_hashes_by_id: dict[str, list[str]] = {}
+ serialized_dags_by_id: dict[str, list[dict[str, Any]]] = {}
+ seen_dag_ids: set[str] = set()
+ nParse = 2 #NOTE: Set parsing number 2, in most of the case twice parse
should catch the stability issues.
+
+ for iteration in range(1, nParse+1):
+ dagbag = _parse_dags_for_stability_check(args.dag_folder)
+
+ dags = dagbag.dags
+ if args.dag_id is not None:
+ dags = {args.dag_id: dagbag.dags[args.dag_id]} if args.dag_id in
dagbag.dags else {}
+
+ for dag_id, dag in sorted(dags.items()):
+ dag_hash, serialized_dag = _serialize_dag_for_stability_check(dag)
+ dag_hashes_by_id.setdefault(dag_id, []).append(dag_hash)
+ serialized_dags_by_id.setdefault(dag_id, []).append(serialized_dag)
+ seen_dag_ids.add(dag_id)
+
+ if args.fail_fast:
+ for dag_id, dag_hashes in dag_hashes_by_id.items():
+ if len(set(dag_hashes)) > 1:
+ break
+ else:
+ continue
+ break
+
+ if args.dag_id is not None and args.dag_id not in seen_dag_ids:
+ raise SystemExit(f"Dag {args.dag_id!r} was not found.")
+
+ unstable_dag_ids = [
+ dag_id for dag_id, dag_hashes in sorted(dag_hashes_by_id.items()) if
len(set(dag_hashes)) > 1
+ ]
+ if unstable_dag_ids:
+ print("Dag stability check failed. The following Dags produced
different serialized output:")
+ for dag_id in unstable_dag_ids:
+ dag_hashes = dag_hashes_by_id[dag_id]
+ print(f"\ndag_id: {dag_id}:")
+ for iteration, dag_hash in enumerate(dag_hashes, start=1):
Review Comment:
The per-Dag header line prints as "dag_id: <id>:" (duplicated label +
trailing colon), which is confusing in the failure output and makes grepping
harder. Consider printing just "<dag_id>:" (or "dag_id: <id>") instead.
##########
airflow-core/src/airflow/cli/cli_config.py:
##########
@@ -1242,6 +1252,21 @@ class GroupCommand(NamedTuple):
func=lazy_load_command("airflow.cli.commands.dag_command.dag_report"),
args=(ARG_BUNDLE_NAME, ARG_OUTPUT, ARG_VERBOSE),
),
+ ActionCommand(
+ name="stability",
+ help="Check that serialized Dags are stable across repeated parses to
prevent dag_version inflection",
+ description=(
Review Comment:
Typo in help text: this should be "dag_version inflation" (the feature this
command is intended to help diagnose), not "inflection".
##########
airflow-core/docs/faq.rst:
##########
@@ -417,7 +417,7 @@ configuration option:
Additionally, you can catch these issues earlier in your development workflow
by using the
`AIR304 <https://docs.astral.sh/ruff/rules/airflow3-dag-dynamic-value/>`_ ruff
rule, which detects
-dynamic values in Dag and Task constructors as part of static linting. See
+dynamic values in Dag and Task constructors as part of static linting or
``airflow dags stability`` CLI command. See
:ref:`best_practices/code_quality_and_linting` for how to set up ruff with
Airflow-specific rules.
Review Comment:
This sentence is grammatically ambiguous: it reads as if the ruff rule
detects issues "as part of ... CLI command". Consider splitting into two
sentences so it's clear the CLI command is an alternative way to catch the same
issues.
##########
airflow-core/src/airflow/cli/commands/dag_command.py:
##########
@@ -79,6 +81,89 @@
# Chunk size for bulk delete.
_RUN_CHUNK_SIZE = 500
+def _normalize_serialized_dag_for_stability_check(serialized_dag: dict[str,
Any]) -> dict[str, Any]:
+ normalized = SerializedDagModel._sort_serialized_dag_dict(serialized_dag)
+ normalized["dag"].pop("fileloc", None)
+ normalized["dag"].pop("bundle_name", None)
+ return normalized
+
+def _serialize_dag_for_stability_check(dag: DAG) -> tuple[str, dict[str, Any]]:
+ serialized_dag = DagSerialization.to_dict(dag)
+ return SerializedDagModel.hash(serialized_dag),
_normalize_serialized_dag_for_stability_check(serialized_dag)
+
+def _format_stability_diff(
+ dag_id: str,
+ first_serialized_dag: dict[str, Any],
+ second_serialized_dag: dict[str, Any],
+) -> str:
+ before = json.dumps(first_serialized_dag, indent=2,
sort_keys=True).splitlines()
+ after = json.dumps(second_serialized_dag, indent=2,
sort_keys=True).splitlines()
+ diff = difflib.unified_diff(
+ before,
+ after,
+ fromfile=f"parse 1: {dag_id} ",
+ tofile=f"parse 2: {dag_id} ",
+ lineterm="",
+ )
+ return "\n".join(diff)
+
+def _parse_dags_for_stability_check(dag_folder: str | None) -> DagBag:
+ return DagBag(dag_folder=dag_folder, safe_mode=True, load_op_links=False)
+
+@cli_utils.action_cli
+@providers_configuration_loaded
+def dag_stability_check(args) -> None:
+ dag_hashes_by_id: dict[str, list[str]] = {}
+ serialized_dags_by_id: dict[str, list[dict[str, Any]]] = {}
+ seen_dag_ids: set[str] = set()
+ nParse = 2 #NOTE: Set parsing number 2, in most of the case twice parse
should catch the stability issues.
+
+ for iteration in range(1, nParse+1):
Review Comment:
Use snake_case for local variables and clean up the inline comment spacing
to match project style (and avoid ruff E261/E262).
--
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]
