From 0d56a126938b570ab2863b72a9d7788bd8df42d6 Mon Sep 17 00:00:00 2001
From: Yoann Abriel <yoann.abriel@gmail.com>
Date: Tue, 10 Mar 2026 15:05:41 +0100
Subject: [PATCH] fix(providers/standard): improve error message for
 non-serializable op_kwargs in PythonVirtualenvOperator

When render_template_as_native_obj=True, templates like {{ ti }} resolve to
live Airflow objects that can't be pickled. The resulting PicklingError from
structlog internals is confusing and doesn't point to the root cause.

Catch serialization failures in _write_args, identify which specific kwargs
failed, and raise a clear AirflowException naming the bad keys and suggesting
to use string attributes instead.

Closes: #61741
---
 .../providers/standard/operators/python.py    | 27 ++++++++++++++++---
 .../unit/standard/operators/test_python.py    | 19 +++++++++++++
 2 files changed, 43 insertions(+), 3 deletions(-)

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.