AWS托管Apache Airflow 2.7.2 API认证及DAG元数据获取问题

一、AWS托管Airflow的API认证方式 #

AWS托管的Apache Airflow（MWAA）的API接口支持两种主流认证方式：

• AWS IAM SigV4签名认证：MWAA原生适配的AWS生态认证方式，通过AWS访问密钥对请求做签名处理，适合AWS内部服务调用场景。
• Airflow API Token（即你说的API Key）认证：生成Airflow专属的Token作为独立凭证，适合非AWS环境下的API调用需求。

二、API Key（Token）的配置步骤 #

1. Airflow（MWAA）端配置 #

要启用Token认证并生成有效凭证，按以下步骤操作：

• 登录MWAA的Airflow UI，点击右上角个人头像，选择Profile进入个人设置页。
• 切换到API Tokens标签，点击Create Token，输入Token名称、可选设置过期时间，确认生成。
• 注意：生成的Token仅显示一次，务必复制保存，丢失后无法找回，只能重新生成。
• 补充配置（若默认未开启）：如果你的MWAA环境未启用Token认证，需要更新环境的airflow.cfg参数：

  api_enabled = True
  auth_backends = airflow.api.auth.backend.token_auth,airflow.api.auth.backend.session
  

  该配置需通过MWAA环境更新操作完成，更新后重启环境生效。

2. POSTMAN中的配置 #

• 新建请求，填入MWAA的Airflow API端点（比如获取DAG元数据的地址：https://<你的MWAA WebServer地址>/api/v1/dags）。
• 切换到Authorization标签，认证类型选择Bearer Token。
• 在Token输入框粘贴之前生成的Airflow API Token，直接发送请求即可获取数据。

3. Python脚本中的配置 #

直接在请求头中携带Bearer Token即可，示例代码：

import requests

# 替换为你的实际地址和Token
MWAA_API_URL = "https://<你的MWAA WebServer地址>/api/v1/dags"
AIRFLOW_API_TOKEN = "<你的Airflow API Token>"

headers = {"Authorization": f"Bearer {AIRFLOW_API_TOKEN}"}
response = requests.get(MWAA_API_URL, headers=headers)

# 验证请求状态
response.raise_for_status()
dag_metadata = response.json()
print(dag_metadata)

如果要用AWS IAM认证调用，可借助requests-aws4auth库处理签名，示例代码：

import requests
from requests_aws4auth import AWS4Auth

AWS_ACCESS_KEY = "<你的AWS访问密钥ID>"
AWS_SECRET_KEY = "<你的AWS秘密访问密钥>"
REGION = "<MWAA所在区域，比如us-east-1>"
SERVICE = "airflow"

auth = AWS4Auth(AWS_ACCESS_KEY, AWS_SECRET_KEY, REGION, SERVICE)
MWAA_API_URL = "https://<你的MWAA WebServer地址>/api/v1/dags"

response = requests.get(MWAA_API_URL, auth=auth)
response.raise_for_status()
print(response.json())

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