ephraimbuddy commented on code in PR #72242:
URL: https://github.com/apache/airflow/pull/72242#discussion_r3887358826
##########
airflow-core/src/airflow/serialization/definitions/param.py:
##########
@@ -82,6 +83,44 @@ def dump(self) -> dict[str, Any]:
}
+class SerializedDagParam:
+ """
+ Scheduler-side DagParam: a late-bound name, not a schema Param.
+
+ ``resolve()`` matches SDK ``DagParam.resolve``: ``dag_run.conf``, then the
+ serialized default, then ``context["params"]``.
+ """
+
+ def __init__(self, *, dag_id: str, name: str, default: Any = NOTSET):
+ self.dag_id = dag_id
+ self.name = name
+ self.default = default
+
+ def iter_references(self):
+ return ()
+
+ def resolve(self, context: Mapping[str, Any]) -> Any:
Review Comment:
`resolve()` is unreachable. Nothing on the scheduler or API side resolves
values out of a deserialized Dag — runtime resolution happens in the Task SDK
against the real `DagParam`, because workers parse the Dag file themselves. The
deliberate precedent is right next door: `SchedulerXComArg` in
`serialization/definitions/xcom_arg.py` has no `resolve()` at all, for exactly
this reason.
`iter_references()` is dead too — `SchedulerXComArg.iter_xcom_references`
dispatches on `ReferenceMixin`, which `SerializedDagParam` does not subclass,
so it is never reached.
Could we drop both methods along with the three
`test_serialized_dagparam_resolve_*` tests? A hand-copied mirror of
`DagParam.resolve` that nothing calls will drift from the SDK silently, and the
tests only prove the copy matches itself.
---
Drafted-by: Claude Code (Opus 5); reviewed by @ephraimbuddy before posting
##########
airflow-core/src/airflow/serialization/serialized_objects.py:
##########
@@ -1100,7 +1110,10 @@ def _serialize_node(cls, op: SdkOperator) -> dict[str,
Any]:
)
value = getattr(op, template_field, None)
if not cls._is_excluded(value, template_field, op):
- serialize_op[template_field] =
serialize_template_field(value, template_field)
+ if isinstance(value, (DagParam, SerializedDagParam)):
Review Comment:
This branch is a behaviour change, not part of the fix. On `main` a
non-mapped `bash_command=dag.param("cmd", "echo hi")` already serializes
stably, because `serialize_template_field` picks up the `serialize()` method:
```
main: "bash_command": {"dag_id": "probe", "default": "echo hi", "name":
"cmd"}
PR: "bash_command": {"__var": {...}, "__type": "dag_param"}
```
There was never a memory address on this path. Changing it rewrites the blob
for every existing Dag that uses `dag.param()` in a template field — one new
DagVersion on upgrade, which is the thing this PR exists to prevent — and flips
the deserialized value from `dict` to `SerializedDagParam`.
The typed form is arguably better, so I am not asking to revert it outright;
I am asking whether it is intended. If it stays, please say so in the PR
description and widen the newsfragment, which currently only mentions mapped
`.partial()`.
---
Drafted-by: Claude Code (Opus 5); reviewed by @ephraimbuddy before posting
##########
airflow-core/src/airflow/serialization/definitions/param.py:
##########
@@ -82,6 +83,44 @@ def dump(self) -> dict[str, Any]:
}
+class SerializedDagParam:
Review Comment:
Minor: no `__eq__`, so two structurally identical params compare unequal.
The siblings in this package use `attrs.define`, which would give you `__eq__`
and `__repr__` for free.
---
Drafted-by: Claude Code (Opus 5); reviewed by @ephraimbuddy before posting
##########
airflow-core/src/airflow/serialization/serialized_objects.py:
##########
@@ -1160,7 +1173,8 @@ def populate_operator(
v = v_in # surpass PLW2901
# Use centralized field deserialization logic
if k in encoded_op.get("template_fields", []):
- pass # Template fields are handled separately
+ if isinstance(v, dict) and v.get(Encoding.TYPE) ==
DAT.DAG_PARAM and Encoding.VAR in v:
Review Comment:
Only a top-level DagParam in a template field gets the typed encoding, so
nesting behaves differently depending on whether the task is mapped:
```python
BashOperator(task_id="nested", env={"A": dag.param("e", "ev")})
BashOperator.partial(task_id="mnested", env={"A": dag.param("m",
"mv")}).expand(...)
```
```
restored non-mapped env: {'A': {'dag_id': 'p2', 'default': 'ev', 'name':
'e'}}
restored mapped env : {'A': SerializedDagParam(dag_id='p2', name='m')}
```
Same author-level construct, two deserialized types. Neither is unstable, so
this is not a correctness bug today, but it is the kind of asymmetry that bites
later.
---
Drafted-by: Claude Code (Opus 5); reviewed by @ephraimbuddy before posting
##########
airflow-core/src/airflow/serialization/serialized_objects.py:
##########
@@ -611,6 +611,10 @@ def serialize(
return cls._encode(cls._serialize_param(var), type_=DAT.PARAM)
elif isinstance(var, XComArg):
return cls._encode(serialize_xcom_arg(var), type_=DAT.XCOM_REF)
+ elif isinstance(var, (DagParam, SerializedDagParam)):
Review Comment:
Is there a path that re-serializes an already-deserialized Dag? I could not
find one, which would make the `SerializedDagParam` half of this isinstance
dead. Happy to be wrong if you know of a caller.
---
Drafted-by: Claude Code (Opus 5); reviewed by @ephraimbuddy before posting
##########
airflow-core/tests/unit/serialization/test_dag_serialization.py:
##########
@@ -5128,3 +5128,172 @@ def get_weight(self, ti):
op = BaseOperator(task_id="empty_task",
weight_rule=NotRegisteredPriorityWeightStrategy())
with pytest.raises(ValueError, match="Unknown priority strategy"):
OperatorSerialization.serialize(op)
+
+
+def _encoded_dag_params(obj):
+ found = []
+ if isinstance(obj, dict):
+ type_ = obj.get(Encoding.TYPE, obj.get("__type"))
+ if type_ in (DagAttributeTypes.DAG_PARAM, "dag_param"):
+ found.append(obj)
+ for value in obj.values():
+ found.extend(_encoded_dag_params(value))
+ elif isinstance(obj, list):
+ for value in obj:
+ found.extend(_encoded_dag_params(value))
+ return found
+
+
+def _taskflow_mapped_dag_with_param(*, default="p_default_val"):
+ from airflow.sdk import task
+
+ with DAG("test-dagparam-mapped", schedule=None, start_date=datetime(2020,
1, 1)) as dag:
+
+ @task
+ def add(value):
+ return value
+
+ add.partial(value=dag.param("p", default)).expand(value=[1, 2, 3])
+ return dag
+
+
+def test_dagparam_in_taskflow_partial_is_serialized_stably():
+ first = DagSerialization.to_dict(_taskflow_mapped_dag_with_param())
+ second = DagSerialization.to_dict(_taskflow_mapped_dag_with_param())
+ blob = json.dumps(first)
+ assert "object at 0x" not in blob.lower()
+ encoded = _encoded_dag_params(first)
+ assert encoded
+ for item in encoded:
+ var = item[Encoding.VAR]
+ assert item[Encoding.TYPE] == DagAttributeTypes.DAG_PARAM
+ assert var["name"] == "p"
+ assert var["dag_id"] == "test-dagparam-mapped"
+ assert var["default"] == "p_default_val"
+ assert first == second
+
+
+def test_dagparam_in_taskflow_partial_roundtrip():
+ serialized = DagSerialization.to_dict(_taskflow_mapped_dag_with_param())
+ restored = DagSerialization.from_dict(serialized)
+ value = restored.task_dict["add"].partial_kwargs["op_kwargs"]["value"]
+ assert isinstance(value, SerializedDagParam)
+ assert value.name == "p"
+ assert value.default == "p_default_val"
+ assert value.dag_id == "test-dagparam-mapped"
+
+
+def test_dagparam_in_mapped_operator_partial():
+ with DAG("test-dagparam-mapped-op", schedule=None,
start_date=datetime(2020, 1, 1)) as dag:
+ MockOperator.partial(task_id="t", arg1=dag.param("p",
"from_partial")).expand(arg2=["a", "b"])
+
+ serialized = DagSerialization.to_dict(dag)
+ assert "object at 0x" not in json.dumps(serialized).lower()
+ restored = DagSerialization.from_dict(serialized)
+ arg1 = restored.task_dict["t"].partial_kwargs["arg1"]
+ assert isinstance(arg1, SerializedDagParam)
+ assert arg1.name == "p"
+ assert arg1.default == "from_partial"
+
+
+def test_dagparam_in_non_mapped_operator_field():
+ with DAG("test-dagparam-plain", schedule=None, start_date=datetime(2020,
1, 1)) as dag:
+ MockOperator(task_id="t", arg1=dag.param("subject", "Hi from
Airflow!"))
+
+ serialized = DagSerialization.to_dict(dag)
+ assert "object at 0x" not in json.dumps(serialized).lower()
+ restored = DagSerialization.from_dict(serialized)
+ arg1 = restored.task_dict["t"].arg1
+ assert isinstance(arg1, SerializedDagParam)
+ assert arg1.name == "subject"
+ assert arg1.default == "Hi from Airflow!"
+
+
+def test_dagparam_notset_default_is_not_stringified():
+ with DAG("test-dagparam-notset", schedule=None, start_date=datetime(2020,
1, 1)) as dag:
+ param = dag.param("p")
+ MockOperator.partial(task_id="t", arg1=param).expand(arg2=[1])
+
+ encoded = BaseSerialization.serialize(param, strict=True)
+ assert encoded[Encoding.TYPE] == DagAttributeTypes.DAG_PARAM
+ assert encoded[Encoding.VAR]["default"][Encoding.TYPE] ==
DagAttributeTypes.ARG_NOT_SET
+
+ restored = DagSerialization.from_dict(DagSerialization.to_dict(dag))
+ arg1 = restored.task_dict["t"].partial_kwargs["arg1"]
+ assert isinstance(arg1, SerializedDagParam)
+ assert arg1.default is NOTSET
+ assert arg1.default != "NOTSET"
+
+
+def test_dagparam_jinja_string_in_partial_stays_string():
+ with DAG("test-dagparam-jinja", schedule=None, start_date=datetime(2020,
1, 1)) as dag:
+ MockOperator.partial(task_id="t", arg1="{{ params.p
}}").expand(arg2=[1])
+
+ serialized = DagSerialization.to_dict(dag)
+ assert _encoded_dag_params(serialized) == []
+ restored = DagSerialization.from_dict(serialized)
+ assert restored.task_dict["t"].partial_kwargs["arg1"] == "{{ params.p }}"
+
+
+def test_two_dagparams_in_one_partial():
+ from airflow.sdk import task
+
+ with DAG("test-dagparam-two", schedule=None, start_date=datetime(2020, 1,
1)) as dag:
+
+ @task
+ def add(left, right, extra):
+ return left, right, extra
+
+ add.partial(left=dag.param("left", "L"), right=dag.param("right",
"R")).expand(extra=[1, 2])
+
+ restored = DagSerialization.from_dict(DagSerialization.to_dict(dag))
+ op_kwargs = restored.task_dict["add"].partial_kwargs["op_kwargs"]
+ assert isinstance(op_kwargs["left"], SerializedDagParam)
+ assert isinstance(op_kwargs["right"], SerializedDagParam)
+ assert op_kwargs["left"].name == "left"
+ assert op_kwargs["right"].name == "right"
+
+
+def test_serialized_dagparam_resolve_prefers_dag_run_conf():
+ param = SerializedDagParam(dag_id="d", name="p", default="from_default")
+ context = {
+ "dag_run": type("DR", (), {"conf": {"p": "from_conf"}})(),
+ "params": {"p": "from_params"},
+ }
+ assert param.resolve(context) == "from_conf"
+ context["dag_run"].conf = {}
+ assert param.resolve(context) == "from_default"
+ param_notset = SerializedDagParam(dag_id="d", name="p")
+ assert param_notset.resolve(
+ {"dag_run": type("DR", (), {"conf": {}})(), "params": {"p":
"from_params"}}
+ ) == ("from_params")
+
+
+def test_serialized_dagparam_resolve_skips_conf_when_name_missing():
+ param = SerializedDagParam(dag_id="d", name="p", default="from_default")
+ context = {
+ "dag_run": type("DR", (), {"conf": {"other": "x"}})(),
+ "params": {"p": "from_params"},
+ }
+ assert param.resolve(context) == "from_default"
+
+
+def test_serialized_dagparam_resolve_raises_when_unresolved():
+ param = SerializedDagParam(dag_id="d", name="p")
+ with pytest.raises(RuntimeError, match="No value could be resolved for
parameter p"):
+ param.resolve({"dag_run": type("DR", (), {"conf": {}})(), "params":
{}})
+
+
+def test_dagparam_nested_in_taskflow_call_is_address_stable():
Review Comment:
This one passes on `main` — `do(dag.param(...))` on a non-mapped `@task`
routes through `serialize_template_field`, so that line of the issue's repro
was never broken. Fine to keep as a regression guard, but the PR description's
"6 failed for the right reason" overstates what is actually pinned;
`test_dagparam_in_non_mapped_operator_field` is similar, in that its `"object
at 0x"` assertion is already true on `main` and only the `isinstance(...,
SerializedDagParam)` assertion is new.
---
Drafted-by: Claude Code (Opus 5); reviewed by @ephraimbuddy before posting
##########
airflow-core/tests/unit/serialization/test_dag_serialization.py:
##########
@@ -5128,3 +5128,172 @@ def get_weight(self, ti):
op = BaseOperator(task_id="empty_task",
weight_rule=NotRegisteredPriorityWeightStrategy())
with pytest.raises(ValueError, match="Unknown priority strategy"):
OperatorSerialization.serialize(op)
+
+
+def _encoded_dag_params(obj):
Review Comment:
Both fallbacks here are redundant: `Encoding.TYPE` is `"__type"` and
`DagAttributeTypes.DAG_PARAM == "dag_param"`, both being str enums.
`obj.get(Encoding.TYPE) == DagAttributeTypes.DAG_PARAM` is enough.
---
Drafted-by: Claude Code (Opus 5); reviewed by @ephraimbuddy before posting
--
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]
