[AIRFLOW-1740] Fix xcom creation and update via RBAC UI (master)

airflow/models/xcom.py

@@ -102,19 +102,7 @@ def set(
         """
         session.expunge_all()
 
-        enable_pickling = configuration.getboolean('core', 'enable_xcom_pickling')
-        if enable_pickling:
-            value = pickle.dumps(value)
-        else:
-            try:
-                value = json.dumps(value).encode('UTF-8')
-            except ValueError:
-                log = LoggingMixin().log
-                log.error("Could not serialize the XCOM value into 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
+        value = XCom.serialize_value(value)
 
         # remove any duplicate XComs
         session.query(cls).filter(
@@ -229,3 +217,20 @@ def delete(cls, xcoms, session=None):
                 )
             session.delete(xcom)
         session.commit()
+
+    @staticmethod
+    def serialize_value(value):
+        # TODO: "pickling" has been deprecated and JSON is preferred.
+        # "pickling" will be removed in Airflow 2.0.
+        if configuration.getboolean('core', 'enable_xcom_pickling'):
+            return pickle.dumps(value)
+
+        try:
+            return json.dumps(value).encode('UTF-8')
+        except ValueError:
+            log = LoggingMixin().log
+            log.error("Could not serialize the XCOM value into 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

airflow/www/views.py

@@ -2043,13 +2043,28 @@ class XComModelView(AirflowModelView):
 
     base_filters = [['dag_id', DagFilter, lambda: []]]
 
+    formatters_columns = {
+        'task_id': wwwutils.task_instance_link,
+        'execution_date': wwwutils.datetime_f('execution_date'),
+        'timestamp': wwwutils.datetime_f('timestamp'),
+        'dag_id': wwwutils.dag_link,
+    }
+
     @action('muldelete', 'Delete', "Are you sure you want to delete selected records?",
             single=False)
     def action_muldelete(self, items):
         self.datamodel.delete_all(items)
         self.update_redirect()
         return redirect(self.get_redirect())
 
+    def pre_add(self, item):
+        item.execution_date = timezone.make_aware(item.execution_date)
+        item.value = XCom.serialize_value(item.value)
+
+    def pre_update(self, item):
+        item.execution_date = timezone.make_aware(item.execution_date)
+        item.value = XCom.serialize_value(item.value)
+
 
 class ConnectionModelView(AirflowModelView):
     route_base = '/connection'