Merge pull request apache#1328 from withnale/www_xcom_view

Added the ability to view XCom variables in webserver

airflow/www/app.py

@@ -1,3 +1,17 @@
+# -*- coding: utf-8 -*-
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
 import socket
 
 from flask import Flask

airflow/www/blueprints.py

@@ -1,3 +1,17 @@
+# -*- coding: utf-8 -*-
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
 from flask import (
     url_for, Markup, Blueprint, redirect,
 )

airflow/www/forms.py

@@ -1,3 +1,17 @@
+# -*- coding: utf-8 -*-
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
 from __future__ import absolute_import
 from __future__ import division
 from __future__ import print_function

airflow/www/templates/airflow/task_instance.html

@@ -1,3 +1,18 @@
+{#
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+#}
 {% extends "airflow/dag.html" %}
 
 {% block head_css %}
@@ -31,6 +46,9 @@ <h4>
     <li><a href="{{ url_for("airflow.log", dag_id=dag.dag_id, task_id=task_id, execution_date=execution_date) }}">
         <span class="glyphicon glyphicon-certificate" aria-hidden="true"></span>
       Log</a></li>
+    <li><a href="{{ url_for("airflow.xcom", dag_id=dag.dag_id, task_id=task_id, execution_date=execution_date) }}">
+        <span class="glyphicon glyphicon-certificate" aria-hidden="true"></span>
+      XCom</a></li>
   </ul>
   <hr>
 {% endblock %}

airflow/www/templates/airflow/xcom.html

@@ -0,0 +1,37 @@
+{#
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+#}
+{% extends "airflow/task_instance.html" %}
+{% block title %}Airflow - DAGs{% endblock %}
+
+{% block body %}
+    {{ super() }}
+    <h4>{{ title }}</h4>
+    <div>
+        <table class="table table-striped table-bordered">
+            <tr>
+                <th>Key</th>
+                <th>Value</th>
+            </tr>
+            {% for attr, value in attributes %}
+                <tr>
+                    <td>{{ attr }}</td>
+                    <td class='code'>{{ value }}</td>
+                </tr>
+            {% endfor %}
+        </table>
+        {{ html_code|safe }}
+    </div>
+{% endblock %}

airflow/www/utils.py

@@ -1,3 +1,17 @@
+# -*- coding: utf-8 -*-
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
 from future import standard_library
 standard_library.install_aliases()
 from builtins import str

airflow/www/views.py

@@ -57,6 +57,7 @@
 from airflow import settings
 from airflow.exceptions import AirflowException
 from airflow.settings import Session
+from airflow.models import XCom
 
 from airflow.utils.json import json_ser
 from airflow.utils.state import State
@@ -901,6 +902,44 @@ def task(self):
             form=form,
             dag=dag, title=title)
 
+    @expose('/xcom')
+    @login_required
+    @wwwutils.action_logging
+    def xcom(self):
+        dag_id = request.args.get('dag_id')
+        task_id = request.args.get('task_id')
+        # Carrying execution_date through, even though it's irrelevant for
+        # this context
+        execution_date = request.args.get('execution_date')
+        dttm = dateutil.parser.parse(execution_date)
+        form = DateTimeForm(data={'execution_date': dttm})
+        dag = dagbag.get_dag(dag_id)
+        if not dag or task_id not in dag.task_ids:
+            flash(
+                "Task [{}.{}] doesn't seem to exist"
+                " at the moment".format(dag_id, task_id),
+                "error")
+            return redirect('/admin/')
+
+        session = Session()
+        xcomlist = session.query(XCom).filter(
+            XCom.dag_id == dag_id, XCom.task_id == task_id,
+            XCom.execution_date == dttm).all()
+
+        attributes = []
+        for xcom in xcomlist:
+            if not xcom.key.startswith('_'):
+                attributes.append((xcom.key, xcom.value))
+
+        title = "XCom"
+        return self.render(
+            'airflow/xcom.html',
+            attributes=attributes,
+            task_id=task_id,
+            execution_date=execution_date,
+            form=form,
+            dag=dag, title=title)\
+
     @expose('/run')
     @login_required
     @wwwutils.action_logging