This is an automated email from the ASF dual-hosted git repository.
potiuk pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new 486f8368e21 Pass runtime values to example Dag shell commands through
env (#73787)
486f8368e21 is described below
commit 486f8368e2119c20a34943f59f80eefaf7bd1b33
Author: Jarek Potiuk <[email protected]>
AuthorDate: Mon Oct 19 17:31:00 2026 +0200
Pass runtime values to example Dag shell commands through env (#73787)
* Pass runtime values to example Dag shell commands through env
Several example Dags built their shell command text out of values that
are only known at run time: another task's output, or a Dag param
rendered through Jinja. Users copy example Dags into real deployments,
so the pattern they show is the pattern that ships. Keep the command
text constant and hand the value to the command through ``env``
instead, so the shell never parses it as part of the command.
Add a ``check-example-dag-shell-commands`` prek hook that flags a
``bash_command`` (or a ``@task.bash`` return value) built with an
f-string, ``str.format``, ``%`` or ``+`` from a runtime value, or with a
Jinja expression rendering ``xcom_pull``, ``params`` or ``dag_run.conf``
into the command text.
Generated-by: Claude Opus 5
* Pass file_count to the bash decorator example as a string
The decorator example hands the describe_dag_folder XComs straight to env,
and a value that json.loads turned into an int makes the subprocess fail
before the command runs. The BashOperator example is unaffected because its
Jinja templates always render to strings.
Generated-by: Claude Opus 5
* Align the example Dag shell-command hook with the other prek scripts
Python 3.10 support is gone and the other prek scripts pin 3.11, so the
3.10 pin made prek provision an extra interpreter just for this hook. The prose
also follows the project's "Dag" spelling.
Generated-by: Claude Opus 5
---
.pre-commit-config.yaml | 6 +
.../airflow/example_dags/example_dag_decorator.py | 20 +-
.../edge3/example_dags/integration_test.py | 5 +-
.../example_dags/example_bash_decorator.py | 17 +-
.../standard/example_dags/example_bash_operator.py | 13 +-
.../ci/prek/check_example_dag_shell_commands.py | 266 +++++++++++++++++++++
.../prek/test_check_example_dag_shell_commands.py | 137 +++++++++++
7 files changed, 444 insertions(+), 20 deletions(-)
diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml
index 41348f5119b..b32abb06dad 100644
--- a/.pre-commit-config.yaml
+++ b/.pre-commit-config.yaml
@@ -1161,6 +1161,12 @@ repos:
language: python
files: ^.*/example_dags/.*\.py$
pass_filenames: true
+ - id: check-example-dag-shell-commands
+ name: Check example Dags do not build shell commands from runtime
values
+ entry: ./scripts/ci/prek/check_example_dag_shell_commands.py
+ language: python
+ files: ^.*/example_dags/.*\.py$
+ pass_filenames: true
- id: check-apache-license-rat
name: Check if licenses are OK for Apache
entry: ./scripts/ci/prek/check_license.py
diff --git a/airflow-core/src/airflow/example_dags/example_dag_decorator.py
b/airflow-core/src/airflow/example_dags/example_dag_decorator.py
index c4c367f4673..452ad97c4b6 100644
--- a/airflow-core/src/airflow/example_dags/example_dag_decorator.py
+++ b/airflow-core/src/airflow/example_dags/example_dag_decorator.py
@@ -59,20 +59,24 @@ def example_dag_decorator(url: str =
"https://httpbingo.org/get";):
"""
get_ip = GetRequestOperator(task_id="get_ip", url=url)
- @task(multiple_outputs=True)
- def prepare_command(raw_json: dict[str, Any]) -> dict[str, str]:
+ @task
+ def get_external_ip(raw_json: dict[str, Any]) -> str:
external_ip = raw_json["origin"]
try:
- ipaddress.ip_address(external_ip)
- return {
- "command": f"echo 'Seems like today your server executing
Airflow is connected from IP {external_ip}'",
- }
+ return str(ipaddress.ip_address(external_ip))
except ValueError:
raise ValueError(f"Invalid IP address: '{external_ip}'.")
- command_info = prepare_command(get_ip.output)
+ external_ip = get_external_ip(get_ip.output)
- BashOperator(task_id="echo_ip_info", bash_command=command_info["command"])
+ # Pass the value to the command through the environment rather than
formatting it into the
+ # command text, so the shell never interprets it as part of the command.
+ BashOperator(
+ task_id="echo_ip_info",
+ bash_command='echo "Seems like today your server executing Airflow is
connected from IP $EXTERNAL_IP"',
+ env={"EXTERNAL_IP": external_ip},
+ append_env=True,
+ )
example_dag = example_dag_decorator()
diff --git
a/providers/edge3/src/airflow/providers/edge3/example_dags/integration_test.py
b/providers/edge3/src/airflow/providers/edge3/example_dags/integration_test.py
index cff3da34634..cea09293c30 100644
---
a/providers/edge3/src/airflow/providers/edge3/example_dags/integration_test.py
+++
b/providers/edge3/src/airflow/providers/edge3/example_dags/integration_test.py
@@ -135,7 +135,10 @@ with DAG(
@task_group(prefix_group_id=False)
def standard_tasks_group():
classic_bash = BashOperator(
- task_id="classic_bash", bash_command="echo Parameter is {{
params.mapping_count }}"
+ task_id="classic_bash",
+ bash_command='echo "Parameter is $MAPPING_COUNT"',
+ env={"MAPPING_COUNT": "{{ params.mapping_count }}"},
+ append_env=True,
)
empty = EmptyOperator(task_id="not_executed")
diff --git
a/providers/standard/src/airflow/providers/standard/example_dags/example_bash_decorator.py
b/providers/standard/src/airflow/providers/standard/example_dags/example_bash_decorator.py
index a8a2073c646..98d3fb3d892 100644
---
a/providers/standard/src/airflow/providers/standard/example_dags/example_bash_decorator.py
+++
b/providers/standard/src/airflow/providers/standard/example_dags/example_bash_decorator.py
@@ -125,17 +125,22 @@ def example_bash_decorator():
set -e
dag_folder="$AIRFLOW_HOME/dags"
file_count=$(find "$dag_folder" -type f -name '*.py' 2>/dev/null |
wc -l)
- printf '{"dag_folder": "%s", "file_count": %s}\\n' "$dag_folder"
"$file_count"
+ printf '{"dag_folder": "%s", "file_count": "%s"}\\n' "$dag_folder"
"$file_count"
"""
dag_stats = describe_dag_folder()
@task.bash
- def show_dag_folder_stats(folder: str, count: int) -> str:
- return f'echo "found {count} Dag file(s) under {folder}"'
-
- # Each key of the returned dict is available as its own XCom.
- show_dag_folder_stats(folder=dag_stats["dag_folder"],
count=dag_stats["file_count"])
+ def show_dag_folder_stats() -> str:
+ return 'echo "found $FILE_COUNT Dag file(s) under $DAG_FOLDER"'
+
+ # Each key of the returned dict is available as its own XCom. Values
produced by another task
+ # reach the command through the environment, so the shell never parses
them as command text.
+ # Environment values must be strings, which is why describe_dag_folder
emits file_count as one.
+ show_dag_folder_stats.override(
+ env={"DAG_FOLDER": dag_stats["dag_folder"], "FILE_COUNT":
dag_stats["file_count"]},
+ append_env=True,
+ )()
# [END howto_decorator_bash_multiple_outputs]
chain(run_me_loop, run_this)
diff --git
a/providers/standard/src/airflow/providers/standard/example_dags/example_bash_operator.py
b/providers/standard/src/airflow/providers/standard/example_dags/example_bash_operator.py
index e7e40354722..7e45568d0de 100644
---
a/providers/standard/src/airflow/providers/standard/example_dags/example_bash_operator.py
+++
b/providers/standard/src/airflow/providers/standard/example_dags/example_bash_operator.py
@@ -95,13 +95,16 @@ with DAG(
output_processor=json.loads,
)
- # Each key of the pushed dict is available as its own XCom.
+ # Each key of the pushed dict is available as its own XCom. Values
produced by another task
+ # reach the command through the environment, so the shell never parses
them as command text.
show_dag_folder_stats = BashOperator(
task_id="show_dag_folder_stats",
- bash_command=(
- "echo \"found {{ ti.xcom_pull(task_ids='describe_dag_folder',
key='file_count') }}"
- " Dag file(s) under {{
ti.xcom_pull(task_ids='describe_dag_folder', key='dag_folder') }}\""
- ),
+ bash_command='echo "found $FILE_COUNT Dag file(s) under $DAG_FOLDER"',
+ env={
+ "DAG_FOLDER": "{{ ti.xcom_pull(task_ids='describe_dag_folder',
key='dag_folder') }}",
+ "FILE_COUNT": "{{ ti.xcom_pull(task_ids='describe_dag_folder',
key='file_count') }}",
+ },
+ append_env=True,
)
# [END howto_operator_bash_multiple_outputs]
describe_dag_folder >> show_dag_folder_stats >> run_this_last
diff --git a/scripts/ci/prek/check_example_dag_shell_commands.py
b/scripts/ci/prek/check_example_dag_shell_commands.py
new file mode 100755
index 00000000000..badd5e2520f
--- /dev/null
+++ b/scripts/ci/prek/check_example_dag_shell_commands.py
@@ -0,0 +1,266 @@
+#!/usr/bin/env python
+# 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.
+# /// script
+# requires-python = ">=3.11,<3.12"
+# dependencies = [
+# "rich>=13.6.0",
+# ]
+# ///
+"""Check that example Dags do not build shell commands out of runtime values.
+
+Example Dags are copied into real deployments, so the pattern they show is the
pattern users
+ship. A shell command assembled with an f-string, ``str.format`` or ``%`` from
a value that is
+only known at run time -- a task's output, a Dag param, ``dag_run.conf`` --
lets that value
+be interpreted by the shell. The value should reach the command through
``env`` instead, and the
+command text should stay a constant.
+
+The check is static and deliberately conservative:
+
+* ``bash_command=`` on any operator call must be a string literal, a name
bound to a module-level
+ constant, or an f-string that only interpolates module-level ``UPPER_CASE``
constants. Anything
+ else -- a subscript or attribute of a task result, ``.format(...)``, ``%``,
``+`` -- is flagged.
+* A function decorated with ``@task.bash`` may not return an f-string,
``.format(...)``, ``%`` or
+ ``+`` expression that interpolates a parameter which some call site feeds
with another task's
+ output or a ``params`` / ``conf`` template. Literals and context values are
the documented
+ ``@task.bash`` idiom and are not flagged.
+
+* Command text -- a literal, a name bound to one, or the literal parts of an
f-string -- may not
+ use a Jinja expression that renders ``xcom_pull``, ``params`` or
``dag_run.conf`` into the
+ command. Airflow renders the template before the shell runs, so the rendered
value is parsed as
+ command text just as an f-string value would be. Context values such as ``{{
ds }}`` are fine.
+"""
+
+from __future__ import annotations
+
+import ast
+import sys
+from pathlib import Path
+
+from rich.console import Console
+
+console = Console(color_system="standard", width=200)
+
+COMMAND_KEYWORDS = ("bash_command",)
+
+
+def _is_literal_text(node: ast.expr | None) -> bool:
+ """Return ``True`` for a string literal, optionally wrapped in
``textwrap.dedent``."""
+ if isinstance(node, ast.Constant) and isinstance(node.value, str):
+ return True
+ if isinstance(node, ast.Call) and len(node.args) == 1 and not
node.keywords:
+ func = node.func
+ name = func.attr if isinstance(func, ast.Attribute) else getattr(func,
"id", None)
+ return name == "dedent" and _is_literal_text(node.args[0])
+ return False
+
+
+def _literal_text(node: ast.expr | None) -> str:
+ """Return the text of a literal accepted by ``_is_literal_text``."""
+ if isinstance(node, ast.Constant) and isinstance(node.value, str):
+ return node.value
+ if isinstance(node, ast.Call) and node.args:
+ return _literal_text(node.args[0])
+ return ""
+
+
+def _literal_bindings(tree: ast.Module) -> tuple[set[str], dict[str, str]]:
+ """
+ Return the names that only ever hold constant command text, and the
literal text they hold.
+
+ Constant names are ``UPPER_CASE`` names and names that are only ever bound
to a literal.
+ """
+ literal_text: dict[str, str] = {}
+ other_names: set[str] = set()
+ for node in ast.walk(tree):
+ targets: list[ast.expr] = []
+ value: ast.expr | None = None
+ if isinstance(node, ast.Assign):
+ targets, value = node.targets, node.value
+ elif isinstance(node, ast.AnnAssign) and node.value is not None:
+ targets, value = [node.target], node.value
+ for target in targets:
+ if not isinstance(target, ast.Name):
+ continue
+ if _is_literal_text(value):
+ literal_text[target.id] = literal_text.get(target.id, "") +
_literal_text(value)
+ else:
+ other_names.add(target.id)
+ upper_names = {name for name in literal_text.keys() | other_names if
name.isupper()}
+ return upper_names | (literal_text.keys() - other_names), literal_text
+
+
+JINJA_EXTERNAL_MARKERS = ("xcom_pull", "params", "dag_run.conf")
+
+
+def _jinja_external(text: str) -> str | None:
+ """Return the external value a Jinja expression in ``text`` renders into
the command, if any."""
+ for opening in text.split("{{")[1:]:
+ expression = opening.split("}}", 1)[0]
+ for marker in JINJA_EXTERNAL_MARKERS:
+ if marker in expression:
+ return f"Jinja template renders {marker} into the command:
{{{{{expression}}}}}"
+ return None
+
+
+def _is_constant_reference(node: ast.expr, constants: set[str]) -> bool:
+ if isinstance(node, ast.Constant):
+ return True
+ if isinstance(node, ast.Name):
+ return node.id in constants or node.id.isupper()
+ if isinstance(node, ast.Attribute):
+ return node.attr.isupper()
+ return False
+
+
+def _unsafe_command(node: ast.expr, constants: set[str], literal_text:
dict[str, str]) -> str | None:
+ """Return why a ``bash_command`` value is built from a runtime value, or
``None`` if it is not."""
+ if isinstance(node, ast.Constant):
+ return _jinja_external(node.value) if isinstance(node.value, str) else
None
+ if _is_literal_text(node):
+ return _jinja_external(_literal_text(node))
+ if isinstance(node, ast.JoinedStr):
+ for part in node.values:
+ if isinstance(part, ast.FormattedValue) and not
_is_constant_reference(part.value, constants):
+ return f"f-string interpolates {ast.unparse(part.value)!r}"
+ if isinstance(part, ast.Constant) and (reason :=
_jinja_external(str(part.value))):
+ return reason
+ return None
+ if isinstance(node, ast.Name) and node.id in literal_text:
+ if reason := _jinja_external(literal_text[node.id]):
+ return reason
+ if isinstance(node, ast.Call) and isinstance(node.func, ast.Attribute) and
node.func.attr == "format":
+ return "command is built with str.format()"
+ if isinstance(node, ast.BinOp) and isinstance(node.op, (ast.Mod, ast.Add)):
+ if _is_constant_reference(node.left, constants) and
_is_constant_reference(node.right, constants):
+ return None
+ return "command is built with string concatenation or % formatting"
+ if isinstance(node, ast.Name):
+ return None if node.id in constants or node.id.isupper() else
f"command comes from {node.id!r}"
+ if isinstance(node, (ast.Subscript, ast.Attribute, ast.Call)):
+ return f"command comes from {ast.unparse(node)!r}"
+ return None
+
+
+def _is_task_bash(decorator: ast.expr) -> bool:
+ target = decorator.func if isinstance(decorator, ast.Call) else decorator
+ return isinstance(target, ast.Attribute) and target.attr == "bash"
+
+
+EXTERNAL_TEMPLATE_MARKERS = ("params", "conf")
+
+
+def _is_external_argument(node: ast.expr) -> bool:
+ """Return ``True`` when a call argument carries a value the Dag author
does not control.
+
+ That is another task's output (``task.output``, ``result["key"]``,
``some_task()``) or a
+ template reading Dag params or ``dag_run.conf``. Literals, loop variables
and context
+ templates such as ``{{ ds }}`` are not external.
+ """
+ if isinstance(node, (ast.Subscript, ast.Attribute, ast.Call)):
+ return True
+ if isinstance(node, ast.Constant) and isinstance(node.value, str) and "{{"
in node.value:
+ return any(marker in node.value for marker in
EXTERNAL_TEMPLATE_MARKERS)
+ return False
+
+
+def _called_name(call: ast.Call) -> str | None:
+ """Return the task function a call invokes, seeing through
``fn.override(...)(...)``."""
+ func = call.func
+ if isinstance(func, ast.Call) and isinstance(func.func, ast.Attribute) and
func.func.attr == "override":
+ func = func.func.value
+ return func.id if isinstance(func, ast.Name) else None
+
+
+def _externally_fed_parameters(
+ tree: ast.Module, function: ast.FunctionDef | ast.AsyncFunctionDef
+) -> set[str]:
+ """Return the parameters of ``function`` that some call site feeds with an
external value."""
+ positional = [a.arg for a in [*function.args.posonlyargs,
*function.args.args]]
+ fed: set[str] = set()
+ for node in ast.walk(tree):
+ if not isinstance(node, ast.Call) or _called_name(node) !=
function.name:
+ continue
+ for index, argument in enumerate(node.args):
+ if index < len(positional) and _is_external_argument(argument):
+ fed.add(positional[index])
+ for keyword in node.keywords:
+ if keyword.arg and _is_external_argument(keyword.value):
+ fed.add(keyword.arg)
+ return fed
+
+
+def _interpolates_parameter(node: ast.expr, parameters: set[str]) -> str |
None:
+ names = {n.id for n in ast.walk(node) if isinstance(n, ast.Name)}
+ if isinstance(node, (ast.JoinedStr, ast.BinOp)) or (
+ isinstance(node, ast.Call) and isinstance(node.func, ast.Attribute)
and node.func.attr == "format"
+ ):
+ used = sorted(names & parameters)
+ if used:
+ return f"returned command interpolates {', '.join(used)}, fed from
another task, params or conf"
+ return None
+
+
+def check_file(file: Path) -> list[str]:
+ """Return the shell-command errors found in a single example Dag file."""
+ try:
+ tree = ast.parse(file.read_text(), filename=str(file))
+ except SyntaxError as exc:
+ return [f"[red]{file}: could not be parsed: {exc}[/]\n"]
+ constants, literal_text = _literal_bindings(tree)
+ errors: list[str] = []
+ for node in ast.walk(tree):
+ if isinstance(node, ast.Call):
+ for keyword in node.keywords:
+ if keyword.arg in COMMAND_KEYWORDS:
+ reason = _unsafe_command(keyword.value, constants,
literal_text)
+ if reason:
+ errors.append(f"[red]{file}:{keyword.value.lineno}:
{keyword.arg}: {reason}.[/]\n")
+ elif isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)) and any(
+ _is_task_bash(d) for d in node.decorator_list
+ ):
+ parameters = _externally_fed_parameters(tree, node)
+ for returned in ast.walk(node):
+ if isinstance(returned, ast.Return) and returned.value is not
None:
+ reason = _interpolates_parameter(returned.value,
parameters)
+ if not reason and _is_literal_text(returned.value):
+ reason = _jinja_external(_literal_text(returned.value))
+ if reason:
+ errors.append(
+ f"[red]{file}:{returned.lineno}: @task.bash
{node.name}: {reason}.[/]\n"
+ )
+ return errors
+
+
+def main(argv: list[str]) -> int:
+ errors: list[str] = []
+ for argument in argv:
+ errors.extend(check_file(Path(argument)))
+ if not errors:
+ return 0
+ console.print("[red]Found example Dags that build shell commands from
runtime values:[/]\n")
+ for error in errors:
+ console.print(error)
+ console.print(
+ "Keep the command text constant and pass runtime values through
[yellow]env[/], e.g.\n"
+ '[yellow]BashOperator(bash_command=\'echo "$VALUE"\', env={"VALUE":
value}, append_env=True)[/]\n'
+ )
+ return 1
+
+
+if __name__ == "__main__":
+ sys.exit(main(sys.argv[1:]))
diff --git a/scripts/tests/ci/prek/test_check_example_dag_shell_commands.py
b/scripts/tests/ci/prek/test_check_example_dag_shell_commands.py
new file mode 100644
index 00000000000..ff19d2fc7ae
--- /dev/null
+++ b/scripts/tests/ci/prek/test_check_example_dag_shell_commands.py
@@ -0,0 +1,137 @@
+# 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
+
+import pytest
+from check_example_dag_shell_commands import check_file
+
+
+class TestCheckFilePasses:
+ """Constant command text, with runtime values passed through ``env``,
produces no errors."""
+
+ @pytest.mark.parametrize(
+ "code",
+ [
+ pytest.param('BashOperator(task_id="t", bash_command="echo
hello")\n', id="literal"),
+ pytest.param(
+ 'BashOperator(task_id="t", bash_command=\'echo "$X"\',
env={"X": result["x"]})\n',
+ id="value-through-env",
+ ),
+ pytest.param(
+ 'BashOperator(task_id="t", bash_command=\'echo "$X"\',
env={"X": "{{ params.x }}"})\n',
+ id="params-template-through-env",
+ ),
+ pytest.param('BashOperator(task_id="t", bash_command="echo {{ ds
}}")\n', id="context-template"),
+ pytest.param(
+ 'CMD = "echo hi"\nBashOperator(task_id="t",
bash_command=CMD)\n',
+ id="module-constant",
+ ),
+ pytest.param(
+ 'BASE = "/tmp"\nBashOperator(task_id="t", bash_command=f"ls
{BASE}")\n',
+ id="f-string-of-upper-case-constant",
+ ),
+ pytest.param(
+ "import textwrap\n"
+ "def f():\n"
+ ' cmd = textwrap.dedent("""\n echo {{ ds }}\n
""")\n'
+ ' BashOperator(task_id="t", bash_command=cmd)\n',
+ id="local-dedented-literal",
+ ),
+ pytest.param(
+ '@task.bash\ndef run(seconds):\n return f"sleep
{seconds}"\nrun(seconds=3)\n',
+ id="task-bash-fed-a-literal",
+ ),
+ pytest.param(
+ '@task.bash\ndef run(ti_key):\n return f"echo
{ti_key}"\nrun()\n',
+ id="task-bash-context-value",
+ ),
+ ],
+ )
+ def test_no_errors(self, write_python_file, code: str):
+ assert check_file(write_python_file(code)) == []
+
+
+class TestCheckFileFails:
+ """A shell command built from a runtime value produces exactly one
error."""
+
+ @pytest.mark.parametrize(
+ ("code", "expected"),
+ [
+ pytest.param(
+ 'BashOperator(task_id="t", bash_command=result["command"])\n',
+ "command comes from",
+ id="task-output-subscript",
+ ),
+ pytest.param(
+ 'BashOperator(task_id="t", bash_command=f"echo {ip}")\n',
+ "f-string interpolates 'ip'",
+ id="f-string-of-runtime-value",
+ ),
+ pytest.param(
+ 'BashOperator(task_id="t", bash_command="echo
{}".format(ip))\n',
+ "str.format",
+ id="str-format",
+ ),
+ pytest.param(
+ 'BashOperator(task_id="t", bash_command="echo %s" % ip)\n',
+ "concatenation or % formatting",
+ id="percent-format",
+ ),
+ pytest.param(
+ 'BashOperator(task_id="t", bash_command="echo {{ params.x
}}")\n',
+ "renders params",
+ id="params-template-in-command",
+ ),
+ pytest.param(
+ 'BashOperator(task_id="t", bash_command="echo {{
ti.xcom_pull(task_ids=\'a\') }}")\n',
+ "renders xcom_pull",
+ id="xcom-template-in-command",
+ ),
+ pytest.param(
+ 'BashOperator(task_id="t", bash_command="echo {{
dag_run.conf[\'x\'] }}")\n',
+ "renders dag_run.conf",
+ id="conf-template-in-command",
+ ),
+ pytest.param(
+ 'CMD = "echo {{ params.x }}"\nBashOperator(task_id="t",
bash_command=CMD)\n',
+ "renders params",
+ id="params-template-through-constant",
+ ),
+ pytest.param(
+ '@task.bash\ndef show(folder):\n return f"echo
{folder}"\nshow(folder=stats["folder"])\n',
+ "interpolates folder",
+ id="task-bash-fed-task-output",
+ ),
+ pytest.param(
+ "@task.bash\n"
+ "def show(folder):\n"
+ ' return f"echo {folder}"\n'
+ 'show.override(task_id="x")(folder=stats["folder"])\n',
+ "interpolates folder",
+ id="task-bash-override-fed-task-output",
+ ),
+ pytest.param(
+ '@task.bash\ndef show():\n return "echo {{ params.x
}}"\nshow()\n',
+ "renders params",
+ id="task-bash-returns-params-template",
+ ),
+ ],
+ )
+ def test_single_error(self, write_python_file, code: str, expected: str):
+ errors = check_file(write_python_file(code))
+ assert len(errors) == 1
+ assert expected in errors[0]
