如何在Airflow DAG中实现受保护REST端点的认证与数据提取

核心实现逻辑拆分为两个核心步骤：先调用登录接口获取合法access_token，再携带token请求受保护的业务数据接口完成提取，生产环境需要额外处理token缓存、过期重试、敏感信息存储等问题。

前置准备 #

• 提前在Airflow Web UI的Connection模块配置外部API的连接信息，不要在DAG代码中硬编码账号密码：连接类型选HTTP，Login字段填API登录账号，Password字段填登录密码，Extra字段可按需配置登录接口路径、业务接口根域名等参数
• 确保Airflow运行环境已安装requests依赖，推荐直接使用Airflow官方提供的HTTP Provider封装的HttpHook，不用重复造轮子处理请求会话、连接复用逻辑

可直接复用的实现示例 #

注意：不要在DAG解析阶段触发token请求，会导致DAG每次刷新都调用认证接口，打垮下游认证服务
以下是基于PythonOperator+HttpHook的完整DAG实现，符合Airflow最佳实践：

from datetime import datetime, timedelta
from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.providers.http.hooks.http import HttpHook
from requests.exceptions import RequestException, HTTPError

default_args = {
    'owner': 'data_etl',
    'retries': 3,
    'retry_delay': timedelta(minutes=2),
    'retry_exception_types': [RequestException]
}

def get_access_token(ti):
    """调用登录接口获取access_token，通过XCom传递给下游任务"""
    http_hook = HttpHook(http_conn_id='external_protected_api', method='POST')
    login_resp = http_hook.run(
        endpoint='/api/v1/auth/login',
        # 按API要求调整传参格式：form表单用data=，json格式用json=
        data={
            "username": http_hook.login,
            "password": http_hook.password
        },
        headers={"Content-Type": "application/x-www-form-urlencoded"},
        timeout=10
    )
    login_resp.raise_for_status()
    token_info = login_resp.json()
    # 按API实际返回字段取token
    access_token = token_info['access_token']
    ti.xcom_push(key='access_token', value=access_token)

def fetch_biz_data(ti):
    """携带token拉取受保护的业务数据"""
    access_token = ti.xcom_pull(task_ids='get_auth_token', key='access_token')
    if not access_token:
        raise RuntimeError("获取access_token失败，终止数据拉取")
    
    http_hook = HttpHook(http_conn_id='external_protected_api', method='GET')
    data_resp = http_hook.run(
        endpoint='/api/v1/data/order/list',
        # 按API要求调整认证头格式：部分接口不需要Bearer前缀，直接传token即可
        headers={"Authorization": f"Bearer {access_token}"},
        data={"start_time": "{{ ds }}", "end_time": "{{ next_ds }}"},
        timeout=30
    )
    # 捕获401异常触发重试，重新获取token
    if data_resp.status_code == 401:
        raise HTTPError("token已过期，触发重试重新获取")
    data_resp.raise_for_status()
    result = data_resp.json()
    # 后续补充数据清洗、落库/写入数仓/对象存储的逻辑
    print(f"拉取完成，共获取{len(result.get('data', []))}条数据")

with DAG(
    dag_id='extract_protected_external_api_data',
    default_args=default_args,
    start_date=datetime(2024, 1, 1),
    schedule_interval=timedelta(hours=1),
    catchup=False,
    tags=['etl', 'external_api']
) as dag:
    task_get_token = PythonOperator(
        task_id='get_auth_token',
        python_callable=get_access_token
    )

    task_fetch_data = PythonOperator(
        task_id='pull_biz_data',
        python_callable=fetch_biz_data
    )

    task_get_token >> task_fetch_data

如果任务逻辑简单，也可以把获取token和拉取数据的逻辑合并到同一个任务函数中，省去XCom传递token的步骤，减少故障点。

生产环境优化建议 #

• 增加token缓存逻辑：如果DAG调度频率高于token有效期，不要每次任务都调用登录接口，可以将有效token存入Airflow Variable或者Redis缓存，每次请求前先校验缓存token是否过期，过期后再重新调用登录接口，降低认证服务压力
• 敏感信息统一托管：不要将账号、密码、token明文写在代码或日志中，推荐对接企业级密钥管理服务作为Airflow的Secrets Backend，统一存储敏感配置
• 完善异常重试：针对网络波动、token偶发失效、接口限流等场景配置细粒度重试规则，不要对参数错误、权限不足等不可重试的异常做无效重试
• 配置合理的超时时间：所有HTTP请求必须设置超时，避免异常请求长时间占用Airflow Worker资源

常见踩坑点 #

• 认证头格式不匹配：部分API的认证头不是Bearer <token>格式，可能要求直接在query参数传token、或者用Token/JWT作为前缀，需要严格按照API文档调整，否则会持续返回403错误
• 登录传参格式错误：注意区分登录接口是form表单提交还是JSON提交，对应调整data/json参数和Content-Type请求头，否则会一直登录失败
• XCom存储限制：默认XCom存储在Airflow元数据库中，不适合存过长的token或大体积数据，长token场景推荐合并任务逻辑，不要通过XCom传递

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