如何将 Python 函数应用于 Airflow Operator 中的 Airflow XCom_pull 值

如何解决如何将 Python 函数应用于 Airflow Operator 中的 Airflow XCom_pull 值

我有一个要求，我需要通过 Airflow sqlsensor 运算符捕获 XCOM 响应并应用一些 python 命令来更改数据格式。

我的代码如下：-

delta_sql_sensor_task = sqlSensor( 
    task_id= "Verify-Completion-of-Delta-Job",poke_interval= 10,timeout=100,conn_id="lct_statedb",#gobal_id_list="{{ str(ti.xcom_pull(key='id_list',task_ids=['hook_task'])).replace('[','') }}",sql="SELECT COALESCE ( (SELECT count(*)  FROM lct_transformation_history WHERE deltaload_transformation_status is not null AND fullload_transformation_status is not null AND category='sites' AND id in (" + "{{ (ti.xcom_pull(key='id_list',task_ids=['hook_task'])) }}" + ") group by category),0) AS TRIGGER;",trigger_rule='one_Failed',dag=dag)

我从代码 "{{ (ti.xcom_pull(key='id_list',task_ids=['hook_task'])).str() }}" 收到的响应为 ([('13',)],)，但我需要将响应响应更改为 '13'

我尝试过的以下选项如下：-

1. 我尝试将 xcom_pull 响应存储在全局变量 gobal_id_list 中并应用 Python 函数，但没有奏效。
2. 我还尝试在 sql 变量中应用 Python 运算符，如下所示，但没有运气，代码如下：- sql="SELECT COALESCE ( (SELECT count(*) FROM lct_transformation_history WHERE deltaload_transformation_status is not null AND fullload_transformation_status is not null AND category='sites' AND id in (" + "{{ str(ti.xcom_pull(key='id_list','').replace(']','').replace('(','').replace(')','').replace(',','') }}" + ") group by category),。

期待有价值的意见。 谢谢。

解决方法

我已经通过以下解决方案解决了我的问题，因为我的 Xcom 响应在元组和列表中，因此我通过下面给出的代码更改解决了：-

"{{ (ti.xcom_pull(key='id_list',task_ids=['hook_task'])).str()[0][0][0] }}

它给了我期望的结果13。