Add LiteralValue wrapper rendered without templating

airflow/template/templater.py

@@ -17,6 +17,7 @@
 
 from __future__ import annotations
 
+from dataclasses import dataclass
 from typing import TYPE_CHECKING, Any, Collection, Iterable, Sequence
 
 from airflow.utils.helpers import render_template_as_native, render_template_to_string
@@ -29,9 +30,27 @@
     from sqlalchemy.orm import Session
 
     from airflow import DAG
+    from airflow.models.operator import Operator
     from airflow.utils.context import Context
 
 
+@dataclass(frozen=True)
+class LiteralValue(ResolveMixin):
+    """
+    A wrapper for a value that should be rendered as-is, without applying jinja templating to its contents.
+
+    :param value: The value to be rendered without templating
+    """
+
+    value: Any
+
+    def iter_references(self) -> Iterable[tuple[Operator, str]]:
+        return ()
+
+    def resolve(self, context: Context) -> Any:
+        return self.value
+
+
 class Templater(LoggingMixin):
     """
     This renders the template fields of object.

airflow/utils/template.py

@@ -0,0 +1,32 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you 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 annotations
+
+from typing import Any
+
+from airflow.template.templater import LiteralValue
+
+
+def literal(value: Any) -> LiteralValue:
+    """
+    Wrap a value to ensure it is rendered as-is without applying Jinja templating to its contents.
+
+    Designed for use in an operator's template field.
+
+    :param value: The value to be rendered without templating
+    """
+    return LiteralValue(value)

docs/apache-airflow/core-concepts/operators.rst

@@ -158,37 +158,50 @@ See the `Jinja documentation <https://jinja.palletsprojects.com/en/2.11.x/api/#j
 
 Some operators will also consider strings ending in specific suffixes (defined in ``template_ext``) to be references to files when rendering fields. This can be useful for loading scripts or queries directly from files rather than including them into DAG code.
 
-For example, consider a BashOperator which runs a multi-line bash script, this will load the file at ``script.sh`` and use its contents as the value for ``bash_callable``:
+For example, consider a BashOperator which runs a multi-line bash script, this will load the file at ``script.sh`` and use its contents as the value for ``bash_command``:
 
 .. code-block:: python
 
     run_script = BashOperator(
         task_id="run_script",
-        bash_callable="script.sh",
+        bash_command="script.sh",
     )
 
 By default, paths provided in this way should be provided relative to the DAG's folder (as this is the default Jinja template search path), but additional paths can be added by setting the ``template_searchpath`` arg on the DAG.
 
-In some cases you may want to disable template rendering on specific fields or prevent airflow from trying to read template files for a given suffix. Consider the following task:
+In some cases, you may want to exclude a string from templating and use it directly. Consider the following task:
 
 .. code-block:: python
 
     print_script = BashOperator(
         task_id="print_script",
-        bash_callable="cat script.sh",
+        bash_command="cat script.sh",
     )
 
+This will fail with ``TemplateNotFound: cat script.sh``, but we can prevent airflow from treating this value as a reference to a file by wrapping it in :func:`~airflow.util.template.literal`.
+This approach disables the rendering of both macros and files and can be applied to selected nested fields while retaining the default templating rules for the remainder of the content.
 
-This will fail with ``TemplateNotFound: cat script.sh``, but we can prevent airflow from treating this value as a reference to a file by overriding ``template_ext``:
+.. code-block:: python
+
+    fixed_print_script = BashOperator(
+        task_id="fixed_print_script",
+        bash_command=literal("cat script.sh"),
+    )
+
+.. versionadded:: 2.8
+    :func:`~airflow.util.template.literal` was added.
+
+Alternatively, if you want to prevent Airflow from treating a value as a reference to a file, you can override ``template_ext``:
 
 .. code-block:: python
 
     fixed_print_script = BashOperator(
         task_id="fixed_print_script",
-        bash_callable="cat script.sh",
+        bash_command="cat script.sh",
     )
     fixed_print_script.template_ext = ()
 
+
 .. _concepts:templating-native-objects:
 
 Rendering Fields as Native Python Objects

tests/models/test_baseoperator.py

@@ -43,6 +43,7 @@
 from airflow.models.taskinstance import TaskInstance
 from airflow.utils.edgemodifier import Label
 from airflow.utils.task_group import TaskGroup
+from airflow.utils.template import literal
 from airflow.utils.trigger_rule import TriggerRule
 from airflow.utils.types import DagRunType
 from airflow.utils.weight_rule import WeightRule
@@ -277,6 +278,9 @@ def test_trigger_rule_validation(self):
             ),
             # By default, Jinja2 drops one (single) trailing newline
             ("{{ foo }}\n\n", {"foo": "bar"}, "bar\n"),
+            (literal("{{ foo }}"), {"foo": "bar"}, "{{ foo }}"),
+            (literal(["{{ foo }}_1", "{{ foo }}_2"]), {"foo": "bar"}, ["{{ foo }}_1", "{{ foo }}_2"]),
+            (literal(("{{ foo }}_1", "{{ foo }}_2")), {"foo": "bar"}, ("{{ foo }}_1", "{{ foo }}_2")),
         ],
     )
     def test_render_template(self, content, context, expected_output):

tests/template/test_templater.py

@@ -20,7 +20,7 @@
 import jinja2
 
 from airflow.models.dag import DAG
-from airflow.template.templater import Templater
+from airflow.template.templater import LiteralValue, Templater
 from airflow.utils.context import Context
 
 
@@ -60,3 +60,23 @@ def test_render_template(self):
         templater.template_ext = [".txt"]
         rendered_content = templater.render_template(templater.message, context)
         assert rendered_content == "Hello world"
+
+    def test_not_render_literal_value(self):
+        templater = Templater()
+        templater.template_ext = []
+        context = Context()
+        content = LiteralValue("Hello {{ name }}")
+
+        rendered_content = templater.render_template(content, context)
+
+        assert rendered_content == "Hello {{ name }}"
+
+    def test_not_render_file_literal_value(self):
+        templater = Templater()
+        templater.template_ext = [".txt"]
+        context = Context()
+        content = LiteralValue("template_file.txt")
+
+        rendered_content = templater.render_template(content, context)
+
+        assert rendered_content == "template_file.txt"