Airflow HttpSensor如何为每次请求生成唯一X-Request-ID头？

你的问题出在：当前配置里的X-Request-ID是在DAG解析阶段就生成了固定UUID，后续每次poke请求都会复用这个ID，触发端点返回400错误。下面给你两种可行方案：

方案一：自定义HttpSensor子类（推荐） #

直接扩展原生HttpSensor，重写poke方法，每次请求前重新生成UUID。这样既能保留HttpSensor的原有功能，又满足每次请求ID唯一的要求：

import uuid
from airflow.sensors.http_sensor import HttpSensor
from airflow.providers.http.hooks.http import HttpHook

class UniqueIdHttpSensor(HttpSensor):
    def poke(self, context):
        # 每次触发检查时生成新的UUID
        self.headers['X-Request-ID'] = str(uuid.uuid4())
        # 调用父类逻辑发送请求并校验响应
        hook = HttpHook(method=self.method, http_conn_id=self.http_conn_id)
        response = hook.run(self.endpoint, data=self.request_params, headers=self.headers)
        return self.response_check(response)

# 使用自定义传感器
monitor_job = UniqueIdHttpSensor(
    task_id='monitor_job',
    http_conn_id='',
    endpoint='http://some_endpoint',
    request_params={},
    response_check=lambda response: 'SUCCESS' in response.text,
    poke_interval=5,
    dag=dag,
    headers={}  # 初始传空字典，每次poke会自动填充新UUID
)

如果后续要用http_conn_id，只需要在Airflow的连接管理里配置好对应的HTTP连接，把http_conn_id设为连接ID即可，不用硬写URL，非常方便。

方案二：PythonSensor结合requests（可作为正式方案） #

你提到的PythonSensor方案其实完全可以作为正式实现，逻辑清晰且易维护，不用受HttpSensor的限制：

from airflow.sensors.python import PythonSensor
import requests
import uuid

def check_job_status():
    # 每次请求生成新UUID
    headers = {'X-Request-ID': str(uuid.uuid4())}
    response = requests.get('http://some_endpoint', headers=headers)
    # 遇到400直接返回False继续重试
    if response.status_code == 400:
        return False
    return 'SUCCESS' in response.text

monitor_job = PythonSensor(
    task_id='monitor_job',
    python_callable=check_job_status,
    poke_interval=5,
    dag=dag
)

两种方案对比 #

• 自定义HttpSensor：适合需要复用HttpHook的认证、超时等配置的场景，和原生HttpSensor用法一致，兼容性好。
• PythonSensor：更轻量化，代码逻辑直观，适合简单的请求监控场景，不需要继承扩展类。

内容的提问来源于stack exchange，提问作者Shaun