[Python] How do I read the output of expanded SQLExecuteQueryOperator in Airflow?

I have a DAG on Airflow with the following structure:

def parameters_listing():
return list_of_paramters

def log_results(query_output):
logging.info(query_output)

with DAG(
dag_id="take_data_from_expand",
start_date=pendulum.datetime(2025, 6, 20, tz="Europe/Rome"),
schedule=None,
catchup=False,
tags=["example", "xcom"],
) as dag:

execute_query_task = SqlExecuteQueryOperator.partial(
task_id="execute_query_task",
conn_id="my_connection",
sql="folder/file_with_query.sql",
do_xcom_push=True,
split_statements=True,
handler=lambda cursor: cursor.fetchall(),
return_last=False,
show_return_value_in_logs=True
).expand(
parameters=parameters_listing.output
)

log_results = PythonOperator(
task_id="log_results",
python_callable=log_results,
op_args=[execute_query_task.output],
)

execute_query_task >> log_results


The SQLExecuteQueryOperator executes ten SELECT statements with the following structure:

SELECT field1, field2, field3 FROM table_name WHERE field1 LIKE '%param%'


The queries are correctly executed and the expected results is logged on Airflow. But when the log_results task is executed, it fails because of this error:

ValueError: too many values to unpack (expected 2)
​

I think the problem is due to the fact that the outputs of the expanded task are collected individually and aren't put in a list. But I don't understand how to solve the problem.

Continue reading...