在Airflow HttpSensor运算符中使用python字符串替换和xcom_pull

from airflow import DAG
from airflow.operators.http_operator import HttpSensor
from airflow.operators.dummy_operator import DummyOperator
from datetime import datetime

default_args = {
    'owner': 'your_name',
    'start_date': datetime(2022, 1, 1),
    'retries': 1,
    'retry_delay': timedelta(minutes=5)
}

with DAG('http_sensor_example', default_args=default_args, schedule_interval='@daily') as dag:
    # 定义HttpSensor运算符
    http_sensor = HttpSensor(
        task_id='http_sensor_task',
        http_conn_id='http_conn',  # 可以配置一个HTTP连接
        endpoint='/api/{{ task_instance.xcom_pull(task_ids="previous_task_id", key="endpoint") }}',  # 使用字符串替换和xcom_pull获取先前任务的输出值
        request_params={'param1': 'value1'}  # 可选的请求参数
    )
    
    # 定义其他任务
    dummy_task = DummyOperator(task_id='dummy_task')
    
    # 设置任务依赖关系
    http_sensor >> dummy_task