Data access in task_name.output when expanding

[dpro-shc]

Let's say we have two tasks, extract and load where extract returns a list of json objects say, [{"foo":"dan","bar":"shc"},{"foo":"alf","bar":"shc"}], and load expands on the output of extract

tasks:
    extract
      operator: airflow.operators.python.PythonOperator
      python_callable_name: extract_fn
      python_callable_file: /usr/local/airflow/dags/extractLoad/utils/python_callables/bronze_tasks.py

    load:
      operator: airflow.operators.python.PythonOperator
      python_callable_name: load_fn
      python_callable_file: /usr/local/airflow/dags/extractLoad/utils/python_callables/bronze_tasks.py
      expand:
        input: extract.output
      dependencies: [extract]

This works, but it's not ideal. The solution is sort of clunky and the internals are obscured-- I have to make the signature of the load task callable load_fn(input), one argument and unpack the values in the function. I'm proposing this functionality:

tasks:
    extract
      operator: airflow.operators.python.PythonOperator
      python_callable_name: extract_fn
      python_callable_file: /usr/local/airflow/dags/extractLoad/utils/python_callables/bronze_tasks.py

    load:
      operator: airflow.operators.python.PythonOperator
      python_callable_name: load_fn
      python_callable_file: /usr/local/airflow/dags/extractLoad/utils/python_callables/bronze_tasks.py
      expand:
        foo: extract.output.foo
        bar: extract.output.bar
      dependencies: [extract]
      

so that I can leave the signature of the load callable load_fn(foo,bar).

[tatiana]

Hi @dpro-shc,

I initially thought this feature was supported by Airflow dynamic task mapping, but I just realised it is not. I'm still learning about DAG Factory. However, I understand that the DAG factory is an abstraction for representing Airflow DAGs as YAMLs. I see this as a layer on top of this. The risk with us adding this type of behaviour is that we'd add additional overhead to the maintenance of DAG Factory with a custom feature that needs to be more generic.

We just released 0.21.0, where we added Dynamic Task mapping - which would probably make it easier to do what you want:

• Support Task Flow and enhance dynamic task mapping #314
• https://github.com/astronomer/dag-factory/releases/tag/v0.21.0
• https://pypi.org/project/dag-factory/0.21.0/

For now, I am closing this issue. If you feel strongly about this feature and would like to contribute, please, reopen this ticket and open a PR. I would be happy to review the PR, and we can discuss it further.