This is an automated email from the ASF dual-hosted git repository.
jason810496 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new 1c3f78e1d57 Validate a serialized Dag and fill its unset settings from
the Airflow config (#74041)
1c3f78e1d57 is described below
commit 1c3f78e1d57f24bd211e426ffcfbb5f8d1583ce5
Author: Jason(Zhe-You) Liu <[email protected]>
AuthorDate: Mon Oct 5 11:38:11 2026 +0800
Validate a serialized Dag and fill its unset settings from the Airflow
config (#74041)
* Check that a serialized Dag can be stored and loaded
DagSerialization.validate_serialized_dag checks the JSON schema,
deserializes a copy and looks for a cycle in the task graph. It raises
DeserializationError naming the Dag, and for a cycle a task on it. It is
meant for a serialized Dag that no Python code built, such as one a
Lang-SDK runtime returns, so the Python SDK's cycle check never ran on
it. The cycle check runs the shared detect_cycle over the serialized
downstream edges.
* Fill a serialized Dag's unset settings from the Airflow config
A Lang-SDK runtime cannot read the Airflow config, so it leaves out the
Dag settings that a Python Dag reads from it when unset. The new
DagSerialization.fill_config_defaults fills max_active_tasks,
max_active_runs, max_consecutive_failed_dag_runs, catchup and
disable_bundle_versioning in a serialized Dag from the config. A value
the Dag sets is kept.
* Reject malformed task entries in a serialized Dag
The schema put additionalProperties on the tasks array, which JSON
Schema ignores, so a task entry like {} passed validation. Each entry
must now be an operator with a string task_id, and
validate_serialized_dag rejects a repeated task id. A schema error
now names the failing field's path.
Also pin both boolean config defaults to different values in the
fill_config_defaults tests, and cover the unwrapped cause of a
deserialization error.
* Check SDK Dags with the Dag processor's validation in conformance
receive() now calls DagSerialization.fill_config_defaults and
validate_serialized_dag, so the conformance job runs the same checks
as the Dag processor and the config-backed field list lives in one
place.
* Note in the parsing ADR that runtimes may omit config-backed fields
* Serialize the SDK Dag in tests that write a dag_maker Dag
These tests passed the dag_maker proxy of the serialized Dag to
LazyDeserializedDAG.from_dag, which serialized each task as its repr.
The tasks schema now rejects that, so they serialize the SDK Dag.
* Sync the Java SDK copy of the Dag schema
* Sync the Go SDK copy of the Dag schema
* Check a serialized Dag as the SDK Dag is checked and return it
validate_serialized_dag now applies the max_active_runs and catchup rules
the
SDK Dag runs when it is built, to the Dag it deserializes. The cycle check
walks that Dag's tasks, so edges from the legacy _downstream_task_ids key
and
from client_defaults count too. It returns the deserialized Dag.
* Report a schema error's path the same way under every jsonschema version
jsonschema 4.26 writes $.dag.tasks[2]['__type'] where 4.23 writes
$.dag.tasks[2].__type, so the import error and three tests changed with
the installed version. Build the path from the error's absolute path.
---
airflow-core/adr/lang-sdk/0004-dag-parsing.md | 12 +-
airflow-core/src/airflow/serialization/schema.json | 13 +-
.../airflow/serialization/serialized_objects.py | 96 +++++++++
airflow-core/tests/unit/models/test_dagrun.py | 6 +-
.../tests/unit/models/test_serialized_dag.py | 4 +-
.../unit/serialization/test_dag_serialization.py | 215 +++++++++++++++++++++
.../unit/serialization/test_dag_version_diff.py | 1 -
go-sdk/schema/dag-schema.json | 13 +-
java-sdk/sdk/schema/dag-schema.json | 13 +-
.../ci/lang_sdk_serialization/serialize_python.py | 23 +--
ts-sdk/schema/dag-schema.json | 13 +-
11 files changed, 376 insertions(+), 33 deletions(-)
diff --git a/airflow-core/adr/lang-sdk/0004-dag-parsing.md
b/airflow-core/adr/lang-sdk/0004-dag-parsing.md
index d4e85acde89..064f44767d2 100644
--- a/airflow-core/adr/lang-sdk/0004-dag-parsing.md
+++ b/airflow-core/adr/lang-sdk/0004-dag-parsing.md
@@ -178,7 +178,7 @@ Airflow Supervisor Bridge
Language Runtime
### DagFileParsingResult Format
-The language runtime must produce a `DagFileParsingResult` that matches Python
Airflow's DagSerialization format exactly. The Airflow scheduler deserializes
this into its internal model — any divergence causes parsing failures.
+The language runtime must produce a `DagFileParsingResult` that matches Python
Airflow's DagSerialization format exactly, except that it may omit the
config-backed Dag fields marked in the table below, which Airflow fills in from
its config before it loads the Dag. The Airflow scheduler deserializes this
into its internal model — any divergence causes parsing failures.
**Envelope:**
@@ -216,9 +216,11 @@ The language runtime must produce a `DagFileParsingResult`
that matches Python A
| `start_date` | float (epoch) | if set | Unwrapped from `__type`/`__var` |
| `end_date` | float (epoch) | if set | Unwrapped from `__type`/`__var` |
| `tags` | list | if non-empty | Unwrapped from `__type`/`__var` |
-| `catchup` | bool | if `true` | |
-| `max_active_tasks` | int | if non-default | |
-| `max_active_runs` | int | if non-default | |
+| `catchup` | bool | if set | Airflow fills an unset field from its config |
+| `max_active_tasks` | int | if set | Airflow fills an unset field from its
config |
+| `max_active_runs` | int | if set | Airflow fills an unset field from its
config |
+| `max_consecutive_failed_dag_runs` | int | if set | Airflow fills an unset
field from its config |
+| `disable_bundle_versioning` | bool | if set | Airflow fills an unset field
from its config |
**Timetable encoding:**
@@ -404,7 +406,7 @@ uv run validation/serialization/compare.py \
validation/serialization/serialized_java.json
```
-Both share test cases defined in `test_dags.yaml`, ensuring the Java SDK
produces byte-identical output to Python's `DagSerialization.serialize_dag()`
for the same inputs.
+Both share test cases defined in `test_dags.yaml`, ensuring the Java SDK
produces byte-identical output to Python's `DagSerialization.serialize_dag()`
for the same inputs, apart from the config-backed Dag fields it may omit, which
Airflow fills in before it loads the Dag.
## Consequences
diff --git a/airflow-core/src/airflow/serialization/schema.json
b/airflow-core/src/airflow/serialization/schema.json
index 086bccb34bb..6ecea324714 100644
--- a/airflow-core/src/airflow/serialization/schema.json
+++ b/airflow-core/src/airflow/serialization/schema.json
@@ -257,7 +257,18 @@
},
"tasks": {
"type": "array",
- "additionalProperties": { "$ref": "#/definitions/operator" }
+ "items": {
+ "type": "object",
+ "required": [ "__type", "__var" ],
+ "properties": {
+ "__type": { "const": "operator" },
+ "__var": {
+ "type": "object",
+ "required": [ "task_id" ],
+ "properties": { "task_id": { "type": "string" } }
+ }
+ }
+ }
},
"params": {
"type": "array",
diff --git a/airflow-core/src/airflow/serialization/serialized_objects.py
b/airflow-core/src/airflow/serialization/serialized_objects.py
index b58bb2505b0..8a6b2599944 100644
--- a/airflow-core/src/airflow/serialization/serialized_objects.py
+++ b/airflow-core/src/airflow/serialization/serialized_objects.py
@@ -22,6 +22,7 @@ from __future__ import annotations
import collections.abc
import contextlib
+import copy
import datetime
import enum
import itertools
@@ -39,11 +40,14 @@ import attrs
import lazy_object_proxy
import pydantic
from dateutil import relativedelta
+from jsonschema import ValidationError
from pendulum.tz.timezone import FixedTimezone, Timezone
+from airflow._shared.dagnode.cycle import detect_cycle
from airflow._shared.module_loading import qualname
from airflow._shared.timezones.timezone import from_timestamp, parse_timezone,
utcnow
from airflow.callbacks.callback_requests import DagCallbackRequest,
TaskCallbackRequest
+from airflow.configuration import conf
from airflow.exceptions import AirflowException, DeserializationError,
SerializationError
from airflow.models.connection import Connection
from airflow.models.expandinput import SchedulerMappedArgument,
create_expand_input
@@ -296,6 +300,16 @@ def _decode_start_trigger_args(var: dict[str, Any]) ->
StartTriggerArgs:
)
+def _build_json_path(error: ValidationError) -> str:
+ """
+ Return where *error* is in the document, such as ``$.dag.tasks[2].__type``.
+
+ Unlike ``ValidationError.json_path``, whose quoting of a key such as
``__type`` depends on the
+ jsonschema version, this is the same under every version.
+ """
+ return "$" + "".join(f"[{part}]" if isinstance(part, int) else f".{part}"
for part in error.absolute_path)
+
+
class _XComRef(NamedTuple):
"""
Store info needed to create XComArg.
@@ -2131,6 +2145,88 @@ class DagSerialization(BaseSerialization):
# Pass client_defaults directly to deserialize_dag
return cls.deserialize_dag(serialized_obj["dag"], client_defaults)
+ @classmethod
+ def fill_config_defaults(cls, serialized_obj: dict[str, Any]) -> None:
+ """
+ Fill in the Dag settings a serialized Dag leaves unset from the
Airflow config, as a Python Dag does.
+
+ A Lang-SDK runtime cannot read the Airflow config, so it leaves
``max_active_tasks``,
+ ``max_active_runs``, ``max_consecutive_failed_dag_runs``, ``catchup``
and
+ ``disable_bundle_versioning`` out unless the Dag sets them. A value
the Dag sets is kept.
+ *serialized_obj* is changed in place.
+ """
+ dag = serialized_obj.get("dag")
+ if not isinstance(dag, dict):
+ # validate_serialized_dag rejects it.
+ return
+ for key, get, section, option in (
+ ("max_active_tasks", conf.getint, "core",
"max_active_tasks_per_dag"),
+ ("max_active_runs", conf.getint, "core",
"max_active_runs_per_dag"),
+ (
+ "max_consecutive_failed_dag_runs",
+ conf.getint,
+ "core",
+ "max_consecutive_failed_dag_runs_per_dag",
+ ),
+ ("catchup", conf.getboolean, "scheduler", "catchup_by_default"),
+ ("disable_bundle_versioning", conf.getboolean, "dag_processor",
"disable_bundle_versioning"),
+ ):
+ if key not in dag:
+ dag[key] = get(section, option)
+
+ @classmethod
+ def validate_serialized_dag(cls, serialized_obj: dict[str, Any]) ->
SerializedDAG:
+ """
+ Check that a serialized Dag, such as one a Lang-SDK runtime produced,
can be stored and loaded.
+
+ It must match the JSON schema, have unique task ids and deserialize.
Like a Dag built with the SDK,
+ it must not set a ``max_active_runs`` its timetable forbids, nor
``catchup`` without a
+ ``start_date``, and its task graph must have no cycle.
*serialized_obj* is not changed.
+
+ :return: The deserialized Dag.
+ :raises DeserializationError: if it does not.
+ """
+ dag = serialized_obj.get("dag")
+ dag_id = dag.get("dag_id") if isinstance(dag, dict) else None
+ try:
+ cls.validate_schema(serialized_obj)
+ except ValidationError as e:
+ raise DeserializationError(
+ dag_id, f"Dag {dag_id!r} does not match the schema at
{_build_json_path(e)}: {e.message}"
+ ) from e
+ task_ids = collections.Counter(
+ task[Encoding.VAR]["task_id"] for task in
serialized_obj["dag"]["tasks"]
+ )
+ if duplicates := sorted(task_id for task_id, count in task_ids.items()
if count > 1):
+ raise DeserializationError(
+ dag_id, f"Dag {dag_id!r} has more than one task with id {',
'.join(map(repr, duplicates))}"
+ )
+ try:
+ deserialized = cls.from_dict(copy.deepcopy(serialized_obj))
+ except Exception as e:
+ cause = e.__cause__ if isinstance(e, DeserializationError) and
e.__cause__ else e
+ raise DeserializationError(
+ dag_id, f"Dag {dag_id!r} cannot be deserialized:
{type(cause).__name__}: {cause}"
+ ) from e
+ if (
+ limit := deserialized.timetable.active_runs_limit
+ ) is not None and deserialized.max_active_runs > limit:
+ raise DeserializationError(
+ dag_id,
+ f"Dag {dag_id!r} sets max_active_runs
{deserialized.max_active_runs}, "
+ f"but {type(deserialized.timetable).__name__} allows at most
{limit}",
+ )
+ if (
+ deserialized.catchup
+ and deserialized.timetable.can_be_scheduled
+ and not (deserialized.start_date or "start_date" in
deserialized.default_args)
+ ):
+ raise DeserializationError(dag_id, f"Dag {dag_id!r} sets catchup
but no start_date")
+ downstream = {task_id: task.downstream_task_ids for task_id, task in
deserialized.task_dict.items()}
+ if (task_id := detect_cycle(downstream, downstream.__getitem__)) is
not None:
+ raise DeserializationError(dag_id, f"Dag {dag_id!r} has a cycle
through task {task_id!r}")
+ return deserialized
+
class TaskGroupSerialization(BaseSerialization):
"""JSON serializable representation of a task group."""
diff --git a/airflow-core/tests/unit/models/test_dagrun.py
b/airflow-core/tests/unit/models/test_dagrun.py
index 47343eeefae..e85713943d6 100644
--- a/airflow-core/tests/unit/models/test_dagrun.py
+++ b/airflow-core/tests/unit/models/test_dagrun.py
@@ -617,7 +617,8 @@ class TestDagRun:
"test_state_succeeded2": TaskInstanceState.SUCCESS,
}
dag.relative_fileloc = relative_fileloc
- SerializedDagModel.write_dag(LazyDeserializedDAG.from_dag(dag),
bundle_name="dag_maker")
+ dag_maker.dag.relative_fileloc = relative_fileloc
+
SerializedDagModel.write_dag(LazyDeserializedDAG.from_dag(dag_maker.dag),
bundle_name="dag_maker")
session.commit()
dag_run = self.create_dag_run(dag=dag,
task_states=initial_task_states, session=session)
@@ -664,7 +665,8 @@ class TestDagRun:
"test_state_failed2": TaskInstanceState.FAILED,
}
dag.relative_fileloc = relative_fileloc
- SerializedDagModel.write_dag(LazyDeserializedDAG.from_dag(dag),
bundle_name="dag_maker")
+ dag_maker.dag.relative_fileloc = relative_fileloc
+
SerializedDagModel.write_dag(LazyDeserializedDAG.from_dag(dag_maker.dag),
bundle_name="dag_maker")
session.commit()
dag_run = self.create_dag_run(dag=dag,
task_states=initial_task_states, session=session)
diff --git a/airflow-core/tests/unit/models/test_serialized_dag.py
b/airflow-core/tests/unit/models/test_serialized_dag.py
index 7609c57ba01..7b47f70fdf2 100644
--- a/airflow-core/tests/unit/models/test_serialized_dag.py
+++ b/airflow-core/tests/unit/models/test_serialized_dag.py
@@ -885,11 +885,11 @@ class TestSerializedDagModel:
Test that dynamic DAG update gracefully handles case where
SerializedDagModel doesn't exist.
This preserves the null-check fix from PR #56422 and tests the direct
UPDATE path.
"""
- with dag_maker(dag_id="test_missing_serdag", serialized=True,
session=session) as dag:
+ with dag_maker(dag_id="test_missing_serdag", serialized=True,
session=session):
EmptyOperator(task_id="task1")
# Write the DAG first
- lazy_dag = LazyDeserializedDAG.from_dag(dag)
+ lazy_dag = LazyDeserializedDAG.from_dag(dag_maker.dag)
SDM.write_dag(
dag=lazy_dag,
bundle_name="test_bundle",
diff --git a/airflow-core/tests/unit/serialization/test_dag_serialization.py
b/airflow-core/tests/unit/serialization/test_dag_serialization.py
index 279a7bb02a1..ce2c2fbb8a1 100644
--- a/airflow-core/tests/unit/serialization/test_dag_serialization.py
+++ b/airflow-core/tests/unit/serialization/test_dag_serialization.py
@@ -5155,3 +5155,218 @@ class TestWeightRule:
op = BaseOperator(task_id="empty_task",
weight_rule=NotRegisteredPriorityWeightStrategy())
with pytest.raises(ValueError, match="Unknown priority strategy"):
OperatorSerialization.serialize(op)
+
+
+class TestValidateSerializedDag:
+ @staticmethod
+ def _serialize() -> dict:
+ with DAG(dag_id="checked_dag", schedule=None) as dag:
+ BaseOperator(task_id="extract") >> BaseOperator(task_id="load")
+ return DagSerialization.to_dict(dag)
+
+ def test_accepts_a_dag_that_loads(self):
+ data = self._serialize()
+ data["__version"] = 2
+ before = copy.deepcopy(data)
+
+ DagSerialization.validate_serialized_dag(data)
+
+ assert data == before
+
+ @pytest.mark.parametrize(
+ ("change", "error"),
+ [
+ pytest.param(
+ {"max_active_runs": "many"},
+ "Dag 'checked_dag' does not match the schema at
$.dag.max_active_runs: "
+ "'many' is not of type 'number'",
+ id="schema",
+ ),
+ pytest.param(
+ {"timetable": {"__type": "no.such.Timetable", "__var": {}}},
+ "Dag 'checked_dag' cannot be deserialized:
TimetableNotRegistered: ",
+ id="deserialize",
+ ),
+ ],
+ )
+ def test_rejects_a_dag_that_does_not_load(self, change, error):
+ data = self._serialize()
+ data["dag"].update(change)
+
+ with pytest.raises(DeserializationError, match=f"^{re.escape(error)}"):
+ DagSerialization.validate_serialized_dag(data)
+
+ def test_returns_the_deserialized_dag(self):
+ dag = DagSerialization.validate_serialized_dag(self._serialize())
+
+ assert isinstance(dag, SerializedDAG)
+ assert dag.dag_id == "checked_dag"
+ assert dag.task_dict["extract"].downstream_task_ids == {"load"}
+
+ @pytest.mark.parametrize(
+ ("entry", "json_path"),
+ [
+ pytest.param({}, "$.dag.tasks[2]", id="empty"),
+ pytest.param("x", "$.dag.tasks[2]", id="not-an-object"),
+ pytest.param(
+ {"__type": "dag", "__var": {"task_id": "t"}},
"$.dag.tasks[2].__type", id="wrong-type"
+ ),
+ pytest.param({"__type": "operator", "__var": {}},
"$.dag.tasks[2].__var", id="no-task-id"),
+ pytest.param(
+ {"__type": "operator", "__var": {"task_id": 5}},
+ "$.dag.tasks[2].__var.task_id",
+ id="task-id-not-a-string",
+ ),
+ ],
+ )
+ def test_rejects_a_task_entry_that_is_not_an_operator(self, entry,
json_path):
+ data = self._serialize()
+ data["dag"]["tasks"].append(entry)
+
+ with pytest.raises(
+ DeserializationError,
+ match=rf"^Dag 'checked_dag' does not match the schema at
{re.escape(json_path)}: ",
+ ):
+ DagSerialization.validate_serialized_dag(data)
+
+ def test_rejects_a_repeated_task_id(self):
+ data = self._serialize()
+ data["dag"]["tasks"].append(copy.deepcopy(data["dag"]["tasks"][0]))
+
+ with pytest.raises(
+ DeserializationError, match="^Dag 'checked_dag' has more than one
task with id 'extract'$"
+ ):
+ DagSerialization.validate_serialized_dag(data)
+
+ def test_reports_the_cause_of_a_deserialization_error(self):
+ data = self._serialize()
+ data["dag"]["tasks"][0]["__var"]["downstream_task_ids"] = ["ghost"]
+
+ with pytest.raises(
+ DeserializationError, match="^Dag 'checked_dag' cannot be
deserialized: KeyError: 'ghost'$"
+ ):
+ DagSerialization.validate_serialized_dag(data)
+
+ def test_rejects_a_dag_with_a_cycle(self):
+ data = self._serialize()
+ load = next(task for task in data["dag"]["tasks"] if
task["__var"]["task_id"] == "load")
+ load["__var"]["downstream_task_ids"] = ["extract"]
+
+ with pytest.raises(DeserializationError, match="^Dag 'checked_dag' has
a cycle through task 'load'$"):
+ DagSerialization.validate_serialized_dag(data)
+
+ def test_rejects_a_cycle_in_the_legacy_downstream_key(self):
+ data = self._serialize()
+ load = next(task for task in data["dag"]["tasks"] if
task["__var"]["task_id"] == "load")
+ load["__var"]["_downstream_task_ids"] = ["extract"]
+
+ with pytest.raises(DeserializationError, match="^Dag 'checked_dag' has
a cycle through task 'load'$"):
+ DagSerialization.validate_serialized_dag(data)
+
+ @pytest.mark.parametrize(
+ ("change", "error"),
+ [
+ pytest.param(
+ {},
+ "Dag 'checked_dag' sets max_active_runs 16, but
ContinuousTimetable allows at most 1",
+ id="unset",
+ ),
+ pytest.param(
+ {"max_active_runs": 2},
+ "Dag 'checked_dag' sets max_active_runs 2, but
ContinuousTimetable allows at most 1",
+ id="above-the-limit",
+ ),
+ ],
+ )
+ def test_rejects_max_active_runs_above_the_timetable_limit(self, change,
error):
+ data = DagSerialization.to_dict(DAG(dag_id="checked_dag",
schedule="@continuous", max_active_runs=1))
+ del data["dag"]["max_active_runs"]
+ data["dag"].update(change)
+
+ with pytest.raises(DeserializationError,
match=f"^{re.escape(error)}$"):
+ DagSerialization.validate_serialized_dag(data)
+
+ def test_rejects_catchup_without_a_start_date(self):
+ data = DagSerialization.to_dict(DAG(dag_id="checked_dag",
schedule="@daily"))
+ data["dag"]["catchup"] = True
+
+ with pytest.raises(DeserializationError, match="^Dag 'checked_dag'
sets catchup but no start_date$"):
+ DagSerialization.validate_serialized_dag(data)
+
+ @pytest.mark.parametrize(
+ "kwargs",
+ [
+ pytest.param(
+ {
+ "schedule": "@daily",
+ "default_args": {"start_date": datetime(2024, 1, 1,
tzinfo=dt_timezone.utc)},
+ },
+ id="start-date-in-default-args",
+ ),
+ pytest.param({"schedule": None}, id="unscheduled"),
+ ],
+ )
+ def test_accepts_catchup_the_sdk_dag_accepts(self, kwargs):
+ dag = DAG(dag_id="checked_dag", catchup=True, **kwargs)
+
+ assert
DagSerialization.validate_serialized_dag(DagSerialization.to_dict(dag)).catchup
is True
+
+
+class TestFillConfigDefaults:
+ CONFIG = {
+ ("core", "max_active_tasks_per_dag"): "7",
+ ("core", "max_active_runs_per_dag"): "3",
+ ("core", "max_consecutive_failed_dag_runs_per_dag"): "5",
+ ("scheduler", "catchup_by_default"): "True",
+ ("dag_processor", "disable_bundle_versioning"): "False",
+ }
+ FIELDS = (
+ "max_active_tasks",
+ "max_active_runs",
+ "max_consecutive_failed_dag_runs",
+ "catchup",
+ "disable_bundle_versioning",
+ )
+
+ @staticmethod
+ def _serialize() -> dict:
+ return DagSerialization.to_dict(DAG(dag_id="native_dag",
schedule=None))
+
+ @conf_vars(CONFIG)
+ def test_fills_an_unset_setting_from_the_config(self):
+ data = self._serialize()
+ for field in self.FIELDS:
+ del data["dag"][field]
+
+ DagSerialization.fill_config_defaults(data)
+
+ assert {field: data["dag"][field] for field in self.FIELDS} == {
+ "max_active_tasks": 7,
+ "max_active_runs": 3,
+ "max_consecutive_failed_dag_runs": 5,
+ "catchup": True,
+ "disable_bundle_versioning": False,
+ }
+
+ @conf_vars(CONFIG)
+ def test_keeps_a_setting_the_dag_sets(self):
+ data = self._serialize()
+ data["dag"].update(
+ max_active_tasks=16,
+ max_active_runs=16,
+ max_consecutive_failed_dag_runs=0,
+ catchup=False,
+ disable_bundle_versioning=True,
+ )
+ before = copy.deepcopy(data)
+
+ DagSerialization.fill_config_defaults(data)
+
+ assert data == before
+
+ def test_leaves_a_payload_without_a_dag_to_validation(self):
+ data = {"__version": 3}
+
+ DagSerialization.fill_config_defaults(data)
+
+ assert data == {"__version": 3}
diff --git a/airflow-core/tests/unit/serialization/test_dag_version_diff.py
b/airflow-core/tests/unit/serialization/test_dag_version_diff.py
index c3b027b6e31..7f92223f604 100644
--- a/airflow-core/tests/unit/serialization/test_dag_version_diff.py
+++ b/airflow-core/tests/unit/serialization/test_dag_version_diff.py
@@ -2896,7 +2896,6 @@ def
test_diff_ignores_wrapper_types_missing_on_both_sides(include_values, proven
base["dag"]["tasks"][0].pop("__type")
base["dag"].pop("task_group")
target = copy.deepcopy(base)
- DagSerialization.validate_schema(base)
assert DagSerialization.from_dict(copy.deepcopy(base)).task_dict == {}
target_version = "new" if provenance_changed else "old"
diff --git a/go-sdk/schema/dag-schema.json b/go-sdk/schema/dag-schema.json
index 086bccb34bb..6ecea324714 100644
--- a/go-sdk/schema/dag-schema.json
+++ b/go-sdk/schema/dag-schema.json
@@ -257,7 +257,18 @@
},
"tasks": {
"type": "array",
- "additionalProperties": { "$ref": "#/definitions/operator" }
+ "items": {
+ "type": "object",
+ "required": [ "__type", "__var" ],
+ "properties": {
+ "__type": { "const": "operator" },
+ "__var": {
+ "type": "object",
+ "required": [ "task_id" ],
+ "properties": { "task_id": { "type": "string" } }
+ }
+ }
+ }
},
"params": {
"type": "array",
diff --git a/java-sdk/sdk/schema/dag-schema.json
b/java-sdk/sdk/schema/dag-schema.json
index 086bccb34bb..6ecea324714 100644
--- a/java-sdk/sdk/schema/dag-schema.json
+++ b/java-sdk/sdk/schema/dag-schema.json
@@ -257,7 +257,18 @@
},
"tasks": {
"type": "array",
- "additionalProperties": { "$ref": "#/definitions/operator" }
+ "items": {
+ "type": "object",
+ "required": [ "__type", "__var" ],
+ "properties": {
+ "__type": { "const": "operator" },
+ "__var": {
+ "type": "object",
+ "required": [ "task_id" ],
+ "properties": { "task_id": { "type": "string" } }
+ }
+ }
+ }
},
"params": {
"type": "array",
diff --git a/scripts/ci/lang_sdk_serialization/serialize_python.py
b/scripts/ci/lang_sdk_serialization/serialize_python.py
index 3bb1406cb0e..0647b53ac37 100644
--- a/scripts/ci/lang_sdk_serialization/serialize_python.py
+++ b/scripts/ci/lang_sdk_serialization/serialize_python.py
@@ -19,8 +19,8 @@ Serialize the Dags of test_dags.yaml with Airflow's own
serializer.
Each Dag is built with the Python authoring API and written as
``DagSerialization.to_dict`` returns it,
keyed by Dag id. ``--receive`` also takes a language SDK's output as Airflow
receives it: it fills in the
-Dag fields the SDK leaves to Airflow's config, writes the result, and loads
every Dag through
-``DagSerialization.validate_schema`` and ``from_dict``. compare.py runs it as::
+Dag fields the SDK leaves to Airflow's config with
``DagSerialization.fill_config_defaults``, writes the
+result, and checks every Dag with
``DagSerialization.validate_serialized_dag``. compare.py runs it as::
uv run --project airflow-core --no-dev python
scripts/ci/lang_sdk_serialization/serialize_python.py \
scripts/ci/lang_sdk_serialization/test_dags.yaml
serialized_python.json \
@@ -30,30 +30,17 @@ Dag fields the SDK leaves to Airflow's config, writes the
result, and loads ever
from __future__ import annotations
import argparse
-import copy
import datetime
import json
import sys
-from collections.abc import Callable
from pathlib import Path
from typing import Any
import yaml
-from airflow.configuration import conf
from airflow.sdk import DAG, BaseOperator, TaskGroup
from airflow.serialization.serialized_objects import DagSerialization
-# The Dag fields the Python DAG reads from config when they are left unset. A
language SDK cannot read
-# Airflow's config, so it leaves them out, and Airflow fills them in when it
receives the SDK's Dags.
-CONFIG_BACKED_DAG_FIELDS: dict[str, Callable[[], Any]] = {
- "max_active_tasks": lambda: conf.getint("core",
"max_active_tasks_per_dag"),
- "max_active_runs": lambda: conf.getint("core", "max_active_runs_per_dag"),
- "max_consecutive_failed_dag_runs": lambda: conf.getint("core",
"max_consecutive_failed_dag_runs_per_dag"),
- "catchup": lambda: conf.getboolean("scheduler", "catchup_by_default"),
- "disable_bundle_versioning": lambda: conf.getboolean("dag_processor",
"disable_bundle_versioning"),
-}
-
class NoopOperator(BaseOperator):
"""Stands in for a language SDK's task: a task with no Python behaviour."""
@@ -102,13 +89,11 @@ def get_node(dag: DAG, groups: dict[str, TaskGroup],
node_id: str):
def receive(sdk_output: Path, received_output: Path) -> None:
received = json.loads(sdk_output.read_text())
for data in received.values():
- for key, read_config in CONFIG_BACKED_DAG_FIELDS.items():
- data["dag"].setdefault(key, read_config())
+ DagSerialization.fill_config_defaults(data)
received_output.write_text(json.dumps(received, indent=2) + "\n")
for dag_id, data in received.items():
try:
- DagSerialization.validate_schema(data)
- DagSerialization.from_dict(copy.deepcopy(data))
+ DagSerialization.validate_serialized_dag(data)
except Exception:
print(f"Airflow cannot load Dag {dag_id!r} as the SDK wrote it",
file=sys.stderr)
raise
diff --git a/ts-sdk/schema/dag-schema.json b/ts-sdk/schema/dag-schema.json
index 086bccb34bb..6ecea324714 100644
--- a/ts-sdk/schema/dag-schema.json
+++ b/ts-sdk/schema/dag-schema.json
@@ -257,7 +257,18 @@
},
"tasks": {
"type": "array",
- "additionalProperties": { "$ref": "#/definitions/operator" }
+ "items": {
+ "type": "object",
+ "required": [ "__type", "__var" ],
+ "properties": {
+ "__type": { "const": "operator" },
+ "__var": {
+ "type": "object",
+ "required": [ "task_id" ],
+ "properties": { "task_id": { "type": "string" } }
+ }
+ }
+ }
},
"params": {
"type": "array",
