Render list items in rendered fields view

airflow/decorators/python.py

@@ -15,7 +15,7 @@
 # specific language governing permissions and limitations
 # under the License.
 
-from typing import Callable, Optional, Sequence, TypeVar
+from typing import Callable, Dict, Optional, Sequence, TypeVar
 
 from airflow.decorators.base import DecoratedOperator, TaskDecorator, task_decorator_factory
 from airflow.operators.python import PythonOperator
@@ -27,15 +27,15 @@ class _PythonDecoratedOperator(DecoratedOperator, PythonOperator):
 
     :param python_callable: A reference to an object that is callable
     :param op_kwargs: a dictionary of keyword arguments that will get unpacked
-        in your function (templated)
+        in your function. (templated)
     :param op_args: a list of positional arguments that will get unpacked when
-        calling your callable (templated)
+        calling your callable. (templated)
     :param multiple_outputs: If set to True, the decorated function's return value will be unrolled to
         multiple XCom values. Dict will unroll to XCom values with its keys as XCom keys. Defaults to False.
     """
 
     template_fields: Sequence[str] = ('op_args', 'op_kwargs')
-    template_fields_renderers = {"op_args": "py", "op_kwargs": "py"}
+    template_fields_renderers: Dict[str, str] = {"op_args": "py", "op_kwargs": "py"}
 
     # since we won't mutate the arguments, we should just do the shallow copy
     # there are some cases we can't deepcopy the objects (e.g protobuf).

airflow/operators/python.py

@@ -106,9 +106,9 @@ def my_python_callable(**kwargs):
 
     :param python_callable: A reference to an object that is callable
     :param op_kwargs: a dictionary of keyword arguments that will get unpacked
-        in your function
+        in your function. (templated)
     :param op_args: a list of positional arguments that will get unpacked when
-        calling your callable
+        calling your callable. (templated)
     :param templates_dict: a dictionary where the values are templates that
         will get templated by the Airflow engine sometime between
         ``__init__`` and ``execute`` takes place and are made available
@@ -122,7 +122,7 @@ def my_python_callable(**kwargs):
     """
 
     template_fields: Sequence[str] = ('templates_dict', 'op_args', 'op_kwargs')
-    template_fields_renderers = {"templates_dict": "json", "op_args": "py", "op_kwargs": "py"}
+    template_fields_renderers: Dict[str, str] = {"templates_dict": "json", "op_args": "py", "op_kwargs": "py"}
     BLUE = '#ffefeb'
     ui_color = BLUE
 

airflow/providers/http/operators/http.py

@@ -41,7 +41,7 @@ class SimpleHttpOperator(BaseOperator):
     :param method: The HTTP method to use, default = "POST"
     :param data: The data to pass. POST-data in POST/PUT and params
         in the URL for a GET request. (templated)
-    :param headers: The HTTP headers to be added to the GET request
+    :param headers: The HTTP headers to be added to the GET request. (templated)
     :param response_check: A check against the 'requests' response object.
         The callable takes the response object as the first positional argument
         and optionally any number of keyword arguments available in the context dictionary.
@@ -61,7 +61,7 @@ class SimpleHttpOperator(BaseOperator):
         'data',
         'headers',
     )
-    template_fields_renderers = {'headers': 'json', 'data': 'py'}
+    template_fields_renderers: Dict[str, str] = {'headers': 'json', 'data': 'py'}
     template_ext: Sequence[str] = ()
     ui_color = '#f4a460'
 

airflow/www/templates/airflow/ti_code.html

@@ -21,10 +21,10 @@
 {% block title %}DAGs - {{ appbuilder.app_name }}{% endblock %}
 
 {% block content %}
-  {{ super() }}
-  <h4>{{ title }}</h4>
-  {% for k, v in html_dict.items() %}
-    <h5>{{ k }}</h5>
-    {{ v }}
-  {% endfor %}
+{{ super() }}
+<h4>{{ title }}</h4>
+{% for k, v in html_dict.items() %}
+<h5><strong>{{ k }}</strong></h5>
+{{ v }}
+{% endfor %}
 {% endblock %}

airflow/www/utils.py

@@ -483,20 +483,37 @@ def pygment_html_render(s, lexer=lexers.TextLexer):
     return highlight(s, lexer(), HtmlFormatter(linenos=True))
 
 
-def render(obj, lexer):
+def render(obj: Any, lexer, handler=None):
     """Render a given Python object with a given Pygments lexer"""
-    out = ""
+
     if isinstance(obj, str):
-        out = Markup(pygment_html_render(obj, lexer))
-    elif isinstance(obj, (tuple, list)):
+        return Markup(pygment_html_render(obj, lexer))
+
+    if isinstance(obj, (tuple, list)):
+        out = ""
         for i, text_to_render in enumerate(obj):
+            if lexer == lexers.PythonLexer:
+                text_to_render = repr(text_to_render)
+
             out += Markup("<div>List item #{}</div>").format(i)
             out += Markup("<div>" + pygment_html_render(text_to_render, lexer) + "</div>")
-    elif isinstance(obj, dict):
+        return out
+
+    if isinstance(obj, dict):
+        out = ""
         for k, v in obj.items():
+            if lexer == lexers.PythonLexer:
+                v = repr(v)
+
             out += Markup('<div>Dict item "{}"</div>').format(k)
             out += Markup("<div>" + pygment_html_render(v, lexer) + "</div>")
-    return out
+        return out
+
+    if obj and handler:
+        return Markup(pygment_html_render(handler(obj), lexer))
+
+    # Return empty string otherwise
+    return ""
 
 
 def json_render(obj, lexer):
@@ -536,8 +553,8 @@ def get_attr_renderer():
         'mysql': lambda x: render(x, lexers.MySqlLexer),
         'postgresql': lambda x: render(x, lexers.PostgresLexer),
         'powershell': lambda x: render(x, lexers.PowerShellLexer),
-        'py': lambda x: render(get_python_source(x), lexers.PythonLexer),
-        'python_callable': lambda x: render(get_python_source(x), lexers.PythonLexer),
+        'py': lambda x: render(x, lexers.PythonLexer, get_python_source),
+        'python_callable': lambda x: render(x, lexers.PythonLexer, get_python_source),
         'rst': lambda x: render(x, lexers.RstLexer),
         'sql': lambda x: render(x, lexers.SqlLexer),
         'tsql': lambda x: render(x, lexers.TransactSqlLexer),

airflow/www/views.py

@@ -1300,12 +1300,9 @@ def rendered_templates(self, session):
         for template_field in task.template_fields:
             content = getattr(task, template_field)
             renderer = task.template_fields_renderers.get(template_field, template_field)
+
             if renderer in renderers:
-                if isinstance(content, (dict, list)):
-                    json_content = json.dumps(content, sort_keys=True, indent=4)
-                    html_dict[template_field] = renderers[renderer](json_content)
-                else:
-                    html_dict[template_field] = renderers[renderer](content)
+                html_dict[template_field] = renderers[renderer](content)
             else:
                 html_dict[template_field] = Markup("<pre><code>{}</pre></code>").format(pformat(content))
 

tests/www/views/test_views_rendered.py

@@ -21,7 +21,9 @@
 import pytest
 
 from airflow.models import DAG, RenderedTaskInstanceFields, Variable
+from airflow.models.baseoperator import BaseOperator
 from airflow.operators.bash import BashOperator
+from airflow.operators.python import PythonOperator
 from airflow.serialization.serialized_objects import SerializedDAG
 from airflow.utils import timezone
 from airflow.utils.session import create_session
@@ -70,6 +72,39 @@ def task_secret(dag):
     )
 
 
+@pytest.fixture()
+def task3(dag):
+    class TestOperator(BaseOperator):
+        template_fields = ('sql',)
+
+        def __init__(self, *, sql, **kwargs):
+            super().__init__(**kwargs)
+            self.sql = sql
+
+        def execute(self, context):
+            pass
+
+    return TestOperator(
+        task_id='task3',
+        sql=['SELECT 1;', 'SELECT 2;'],
+        dag=dag,
+    )
+
+
+@pytest.fixture()
+def task4(dag):
+    def func(*op_args):
+        pass
+
+    return PythonOperator(
+        task_id='task4',
+        python_callable=func,
+        op_args=['{{ task_instance_key_str }}_args'],
+        op_kwargs={'0': '{{ task_instance_key_str }}_kwargs'},
+        dag=dag,
+    )
+
+
 @pytest.fixture(scope="module", autouse=True)
 def init_blank_db():
     """Make sure there are no runs before we test anything.
@@ -82,15 +117,15 @@ def init_blank_db():
 
 
 @pytest.fixture(autouse=True)
-def reset_db(dag, task1, task2, task_secret):
+def reset_db(dag, task1, task2, task3, task4, task_secret):
     yield
     clear_db_dags()
     clear_db_runs()
     clear_rendered_ti_fields()
 
 
 @pytest.fixture()
-def create_dag_run(dag, task1, task2, task_secret):
+def create_dag_run(dag, task1, task2, task3, task4, task_secret):
     def _create_dag_run(*, execution_date, session):
         dag_run = dag.create_dagrun(
             state=DagRunState.RUNNING,
@@ -105,6 +140,11 @@ def _create_dag_run(*, execution_date, session):
         ti2.state = TaskInstanceState.SCHEDULED
         ti3 = dag_run.get_task_instance(task_secret.task_id, session=session)
         ti3.state = TaskInstanceState.QUEUED
+        ti4 = dag_run.get_task_instance(task3.task_id, session=session)
+        ti4.state = TaskInstanceState.SUCCESS
+        ti5 = dag_run.get_task_instance(task4.task_id, session=session)
+        ti5.state = TaskInstanceState.SUCCESS
+
         session.flush()
         return dag_run
 
@@ -205,3 +245,38 @@ def test_rendered_template_secret(admin_client, create_dag_run, task_secret):
     )
     ti.refresh_from_task(task_secret)
     assert ti.state == TaskInstanceState.QUEUED
+
+
+@pytest.mark.usefixtures("patch_app")
+def test_rendered_template_view_for_list_template_field_args(admin_client, create_dag_run, task3):
+    """
+    Test that the Rendered View can show a list of syntax-highlighted SQL statements
+    """
+    assert task3.sql == ['SELECT 1;', 'SELECT 2;']
+
+    with create_session() as session:
+        create_dag_run(execution_date=DEFAULT_DATE, session=session)
+
+    url = f'rendered-templates?task_id=task3&dag_id=testdag&execution_date={quote_plus(str(DEFAULT_DATE))}'
+
+    resp = admin_client.get(url, follow_redirects=True)
+    check_content_in_response("List item #0", resp)
+    check_content_in_response("List item #1", resp)
+
+
+@pytest.mark.usefixtures("patch_app")
+def test_rendered_template_view_for_op_args(admin_client, create_dag_run, task4):
+    """
+    Test that the Rendered View can show rendered values in op_args and op_kwargs
+    """
+    assert task4.op_args == ['{{ task_instance_key_str }}_args']
+    assert list(task4.op_kwargs.values()) == ['{{ task_instance_key_str }}_kwargs']
+
+    with create_session() as session:
+        create_dag_run(execution_date=DEFAULT_DATE, session=session)
+
+    url = f'rendered-templates?task_id=task4&dag_id=testdag&execution_date={quote_plus(str(DEFAULT_DATE))}'
+
+    resp = admin_client.get(url, follow_redirects=True)
+    check_content_in_response('testdag__task4__20200301_args', resp)
+    check_content_in_response('testdag__task4__20200301_kwargs', resp)