Add multiple_outputs option to PythonOperator

[jannisko]

Description

Add a new optional field return_multiple to PythonOperator, which automatically splits returned dicts into separate XComs.

Use case/motivation

Currently, the task-flow decorator @task accepts a boolean argument multiple_outputs, which automatically splits any returned dictionary of values into separate xcom values. Using this feature, the following code can be compacted from

@task
def split_csv(task_instance: TaskInstance, csv_path: ObjectStoragePath) -> None
    ...
    task_instance.xcom_push("rows_csv", rows_csv)
    task_instance.xcom_push("summary_csv", summary_csv)

to

@task(mulitple_outputs=True)
def split_csv(csv_path: ObjectStoragePath) -> dict[str, ObjectStoragePath]
    ...
    return {"rows_csv": rows_csv, "summary_csv": summary_csv}

This abstracts away the airflow-ism of task instances and still enables using each part of the returned value separately via e.g. split_csv.output["rows_csv"].

This feature is not available for any other operator, though I feel like it would be very useful. I'm open to implementing this by applying the existing code from airflow.decorators.base.DecoratedOperator to PythonOperator, since BaseOperator doesn't implement any handling of returned values.

Please let me know what you think about this change. 🙂

Related issues

No response

Are you willing to submit a PR?

• Yes I am willing to submit a PR!

Code of Conduct

• I agree to follow this project's Code of Conduct

[boring-cyborg]

Thanks for opening your first issue here! Be sure to follow the issue template! If you are willing to raise PR to address this issue please do so, no need to wait for approval.

[hussein-awala]

I think it's a good idea, I will take a look.