apache / apache/airflow

Allow template resolution of a upstream string passed via XComArg

Open
#26,016 13 comments 0 reactions 0 assignees View on GitHub
kind:feature
Dominant language
Python
Stars
46.9k
Forks
17.8k
Avg merge
2d 10h
Merged PRs (30d)
483

Description

Edit: The original issue (kept verbatim under the separator) described a feature request toward dynamically mapped tasks specifically, but this non-rendering behaviour is not specific to dynamic task mapping, as explained in https://github.com/apache/airflow/issues/26016#issuecomment-1229364541 below. Please keep in mind this feature is technically orthogonal to dynamic task-mapping when discussing.

-----

### Description

Consider the following pseudocode:

```
@task
def get_templated_echo():
return [
'echo "The date is {{ds}}"',
'echo "The time is {{ts}}"'
]

BashOperator.partial(task_id="echo_task").expand(bash_command=get_templated_echos())
```

This will result in generating two BashOperator tasks that execute echo statements, but the templates won't be rendered, despite `bash_command` being a template field.

The expected result is that the template would be applied to templated keyword arguments even when applied via `expand()`

### Use case/motivation

While the above example can be trivially solved by accessing the dag run context inside the function, let's take about this use case.

I want to allow users to provide any number of '.sql' files with internal template strings to allow semi-technical users to drop in their SELECT statements into a folder as a file and have them be run as part of a standard routine (yes, there's lots of protection against injection). These files would internally rely on templating.

Much simplified example.

file contents: (assume there are a few dozen of these doing different things)
```
SELECT '{{ts_nodash}}' as dag_run_timestamp, 'file1' as filename;
```

Dag works kinda like this:
```
@task
def get_filenames_from_directory(dir):
return [ listoffiles in dir]

SnowflakeOperator.partial(task_id='run_sql_files').expand(get_filenames_from_directory('/opt/airflow/dags/sql/mysubdir'))
```

In fact, we're passing this into a subclass of SnowflakeOperator that validates and cleans the SQL and uses it as part of a CTAS. We're working around this by just running the listdir function on dag load, but would prefer to do it dynamically at runtime

### Related issues

_No response_

### Are you willing to submit a PR?

- [X] Yes I am willing to submit a PR!

### Code of Conduct

- [X] I agree to follow this project's [Code of Conduct](https://github.com/apache/airflow/blob/main/CODE_OF_CONDUCT.md)

Contributor guide

Open the contributing guide

Research direction

The issue's entry points are XComArg values passed through expand(), template fields such as BashOperator.bash_command, and the SnowflakeOperator use case. Trace how expanded arguments reach template rendering, then verify completion by confirming that XComArg-provided Bash and SQL strings render for each mapped task, including values such as ts_nodash.

Written by the indexing model from the issue text.

Assessment

Tech stack
python
Domain
data-engineering
Issue type
Feature
Difficulty
4/5
Estimated time
3-5 days
Activity status
Stale
Clarity
Mostly clear
Newbie friendliness
35/100

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.