Airflow 2.6.3内部调用Rest API跨AWS环境及认证问题求助

一、先解决Basic Auth 401的核心问题 #

Airflow 2.6.3的API默认不开启Basic Auth，默认用Flask-AppBuilder的session认证（也就是你提到的Cookie方式），这就是你外部代码传了账号密码仍返回401的原因。要启用Basic Auth，需修改Airflow配置：

1. 编辑airflow.cfg的[api]段：

   [api]
   auth_backends = airflow.api.auth.backend.basic_auth, airflow.api.auth.backend.session
   

   把basic_auth加到认证后端列表最前面，保存后重启Airflow webserver和scheduler。

2. 重启后再运行你的外部Python代码，就能正常返回200了。

二、Airflow内部调用API的替代方案（取代SimpleHttpOperator） #

SimpleHttpOperator的认证逻辑确实受限，没法直接支持Basic Auth，推荐用PythonOperator写自定义请求逻辑，灵活性更高，还能轻松适配多AWS环境的认证需求：

示例代码（PythonOperator实现跨环境元数据拉取与对比） #

from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime
import requests
from requests.auth import HTTPBasicAuth

def fetch_airflow_metadata(env_config):
    """从指定Airflow环境拉取元数据"""
    url = f"{env_config['base_url']}/api/v1/config"
    try:
        response = requests.get(
            url,
            auth=HTTPBasicAuth(env_config['username'], env_config['password']),
            headers={'Accept': 'application/json'},
            verify=False  # 生产环境建议移除，配置合法SSL证书
        )
        response.raise_for_status()
        return response.json()
    except Exception as e:
        raise Exception(f"拉取{env_config['env_name']}元数据失败: {str(e)}")

def compare_metadata(**context):
    """对比多AWS环境的Airflow元数据"""
    # 定义各环境的Airflow配置（敏感信息用Airflow变量存储）
    envs = [
        {
            "env_name": "生产环境",
            "base_url": "https://prod-airflow.example.com",
            "username": "{{ var.value.prod_airflow_user }}",
            "password": "{{ var.value.prod_airflow_pass }}"
        },
        {
            "env_name": "测试环境",
            "base_url": "https://test-airflow.example.com",
            "username": "{{ var.value.test_airflow_user }}",
            "password": "{{ var.value.test_airflow_pass }}"
        }
    ]
    
    # 拉取所有环境的元数据
    metadata_results = {}
    for env in envs:
        # 渲染Airflow变量（避免硬编码敏感信息）
        rendered_env = context['task_instance'].render_template(env)
        metadata_results[env['env_name']] = fetch_airflow_metadata(rendered_env)
    
    # 核心配置对比逻辑示例
    prod_config = metadata_results['生产环境']
    test_config = metadata_results['测试环境']
    diffs = []
    for config_group in ['core', 'database', 'api']:
        if prod_config.get(config_group) != test_config.get(config_group):
            diffs.append(f"配置组「{config_group}」存在差异")
    
    # 输出差异（可扩展为存入S3、数据库或发送告警）
    if diffs:
        print("环境配置差异:\n" + "\n".join(diffs))
    else:
        print("所有环境核心配置一致")

with DAG(
    dag_id='airflow_multienv_metadata_comparison',
    schedule_interval='@daily',
    start_date=datetime(2024, 1, 1),
    catchup=False
) as dag:
    compare_task = PythonOperator(
        task_id='compare_environment_metadata',
        python_callable=compare_metadata,
        provide_context=True
    )

compare_task

三、额外实用建议 #

• 敏感信息管理: 绝对不要硬编码账号密码，用Airflow的Variable或AWS Secrets Manager存储，示例中已使用Airflow Variable实现。
• 安全优化: 生产环境禁用verify=False，配置合法的SSL证书路径，避免安全风险。
• 权限控制: 给拉取元数据的账号分配Viewer角色即可，遵循最小权限原则，不要使用管理员账号。
• 扩展对比维度: 除了系统配置，还可以拉取DAG列表、任务实例状态、连接配置等元数据，实现更全面的环境差异对比。
• 容错机制: 可以用tenacity库给请求添加重试逻辑，处理临时网络故障或API限流问题。

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