diff --git a/airflow/decorators/python.py b/airflow/decorators/python.py index c7c1a629a4dfb..b72f852994517 100644 --- a/airflow/decorators/python.py +++ b/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). diff --git a/airflow/operators/python.py b/airflow/operators/python.py index 920a75d08f766..813f0c3cfc04c 100644 --- a/airflow/operators/python.py +++ b/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 diff --git a/airflow/providers/http/operators/http.py b/airflow/providers/http/operators/http.py index 0622b8e7d19e9..74302682363b7 100644 --- a/airflow/providers/http/operators/http.py +++ b/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' diff --git a/airflow/www/templates/airflow/ti_code.html b/airflow/www/templates/airflow/ti_code.html index ef5a28de4f6ce..554af641ea39b 100644 --- a/airflow/www/templates/airflow/ti_code.html +++ b/airflow/www/templates/airflow/ti_code.html @@ -21,10 +21,10 @@ {% block title %}DAGs - {{ appbuilder.app_name }}{% endblock %} {% block content %} - {{ super() }} -

{{ title }}

- {% for k, v in html_dict.items() %} -
{{ k }}
- {{ v }} - {% endfor %} +{{ super() }} +

{{ title }}

+{% for k, v in html_dict.items() %} +
{{ k }}
+{{ v }} +{% endfor %} {% endblock %} diff --git a/airflow/www/utils.py b/airflow/www/utils.py index 2bdc2939bfd61..3e50fbfca856a 100644 --- a/airflow/www/utils.py +++ b/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("
List item #{}
").format(i) out += Markup("
" + pygment_html_render(text_to_render, lexer) + "
") - 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('
Dict item "{}"
').format(k) out += Markup("
" + pygment_html_render(v, lexer) + "
") - 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), diff --git a/airflow/www/views.py b/airflow/www/views.py index 48f6a47f5e0d0..4232a5bdc4157 100644 --- a/airflow/www/views.py +++ b/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("
{}
").format(pformat(content)) diff --git a/tests/www/views/test_views_rendered.py b/tests/www/views/test_views_rendered.py index 31935221c0714..d3be908560e4f 100644 --- a/tests/www/views/test_views_rendered.py +++ b/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,7 +117,7 @@ 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() @@ -90,7 +125,7 @@ def reset_db(dag, task1, task2, task_secret): @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)