diff --git a/providers/standard/src/airflow/providers/standard/operators/python.py b/providers/standard/src/airflow/providers/standard/operators/python.py index 153044a5c3f9e..8d800d8fab11d 100644 --- a/providers/standard/src/airflow/providers/standard/operators/python.py +++ b/providers/standard/src/airflow/providers/standard/operators/python.py @@ -579,9 +579,30 @@ def resolve_proxies(obj): if self.op_args or self.op_kwargs: self.log.info("Use %r as serializer.", self.serializer) - file.write_bytes( - self.pickling_library.dumps({"args": self.op_args, "kwargs": resolve_proxies(self.op_kwargs)}) - ) + resolved_kwargs = resolve_proxies(self.op_kwargs) + try: + file.write_bytes( + self.pickling_library.dumps({"args": self.op_args, "kwargs": resolved_kwargs}) + ) + except Exception: + # Identify which specific kwarg(s) failed to serialize so the + # user gets a clear error instead of an opaque PicklingError. + bad_keys: list[str] = [] + for key, value in resolved_kwargs.items(): + try: + self.pickling_library.dumps(value) + except Exception: + bad_keys.append(key) + if bad_keys: + raise AirflowException( + f"Failed to serialize op_kwargs. The following keys contain objects that " + f"cannot be pickled: {bad_keys}. This often happens when " + f"render_template_as_native_obj=True and templates like '{{{{ ti }}}}' " + f"resolve to live Airflow objects instead of strings. Use string " + f"representations or pass only the specific attributes you need " + f"(e.g. '{{{{ ti.task_id }}}}', '{{{{ ti.run_id }}}}')." + ) + raise def _write_string_args(self, file: Path): file.write_text("\n".join(map(str, self.string_args))) diff --git a/providers/standard/tests/unit/standard/operators/test_python.py b/providers/standard/tests/unit/standard/operators/test_python.py index 22497f3646ad2..437ae89ac26ff 100644 --- a/providers/standard/tests/unit/standard/operators/test_python.py +++ b/providers/standard/tests/unit/standard/operators/test_python.py @@ -1092,6 +1092,25 @@ def f(): op = self.opcls(task_id="task", python_callable=f, **self.default_kwargs()) copy.deepcopy(op) + def test_write_args_non_serializable_op_kwargs(self, tmp_path): + """Non-serializable op_kwargs should raise AirflowException with helpful message.""" + + class NonSerializable: + def __reduce__(self): + raise TypeError("cannot pickle this") + + def f(x): + return x + + op = self.opcls( + task_id="task", + python_callable=f, + op_kwargs={"bad_obj": NonSerializable(), "good_obj": "hello"}, + **self.default_kwargs(), + ) + with pytest.raises(AirflowException, match=r"cannot be pickled.*\['bad_obj'\]"): + op._write_args(tmp_path / "args.pkl") + def test_virtualenv_serializable_context_fields(self, create_task_instance): """Ensure all template context fields are listed in the operator.