This is an automated email from the ASF dual-hosted git repository.
kaxilnaik pushed a commit to branch v1-10-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/v1-10-test by this push:
new d911ba0 Add XCom.deserialize_value to Airflow 1.10.13 (#12328)
d911ba0 is described below
commit d911ba07ae12870fe5b3ace79122885113843629
Author: Kaxil Naik <[email protected]>
AuthorDate: Fri Nov 13 00:20:59 2020 +0000
Add XCom.deserialize_value to Airflow 1.10.13 (#12328)
closes https://github.com/apache/airflow/issues/11988
---
airflow/models/xcom.py | 33 +++++++++++++++++++++++----------
1 file changed, 23 insertions(+), 10 deletions(-)
diff --git a/airflow/models/xcom.py b/airflow/models/xcom.py
index ea902be..e33c5e5 100644
--- a/airflow/models/xcom.py
+++ b/airflow/models/xcom.py
@@ -67,17 +67,13 @@ class BaseXCom(Base, LoggingMixin):
"""
@reconstructor
def init_on_load(self):
- enable_pickling = conf.getboolean('core', 'enable_xcom_pickling')
- if enable_pickling:
+ try:
+ self.value = self.deserialize_value(self)
+ except (UnicodeEncodeError, ValueError):
+ # For backward-compatibility.
+ # Preventing errors in webserver
+ # due to XComs mixed with pickled and unpickled.
self.value = pickle.loads(self.value)
- else:
- try:
- self.value = json.loads(self.value.decode('UTF-8'))
- except (UnicodeEncodeError, ValueError):
- # For backward-compatibility.
- # Preventing errors in webserver
- # due to XComs mixed with pickled and unpickled.
- self.value = pickle.loads(self.value)
def __repr__(self):
return '<XCom "{key}" ({task_id} @ {execution_date})>'.format(
@@ -233,6 +229,23 @@ class BaseXCom(Base, LoggingMixin):
"support for XCOM in your airflow config.")
raise
+ @staticmethod
+ def deserialize_value(result):
+ # TODO: "pickling" has been deprecated and JSON is preferred.
+ # "pickling" will be removed in Airflow 2.0.
+ enable_pickling = conf.getboolean('core', 'enable_xcom_pickling')
+ if enable_pickling:
+ return pickle.loads(result.value)
+
+ try:
+ return json.loads(result.value.decode('UTF-8'))
+ except ValueError:
+ log.error("Could not deserialize the XCOM value from JSON. "
+ "If you are using pickles instead of JSON "
+ "for XCOM, then you need to enable pickle "
+ "support for XCOM in your airflow config.")
+ raise
+
def resolve_xcom_backend():
"""Resolves custom XCom class"""
