使用Google OAuth2 id_token时Airflow DAG触发DefaultCredentialsError报错

问题原因 #

你的代码中id_token.fetch_id_token未传入从Airflow连接获取的凭证参数，导致该方法默认去环境变量、本地文件或元数据服务器查找凭证，而非使用你在google_cloud_default中配置的JSON密钥。

修正后的代码 #

from google.auth.transport.requests import Request
from google.auth import id_token
from airflow.providers.google.common.hooks.base_google import GoogleBaseHook

def get_identity_token(service_url):
    # 初始化GCP Hook，指定Airflow中配置的连接ID
    hook = GoogleBaseHook(gcp_conn_id='google_cloud_default')
    # 获取连接对应的服务账号凭证
    creds = hook.get_credentials()

    # 刷新过期的凭证（如果需要）
    if creds.expired and creds.refresh_token:
        creds.refresh(Request())

    # 使用指定的凭证生成目标服务的ID Token
    token = id_token.fetch_id_token(Request(), service_url, credentials=creds)
    return token

额外检查项 #

• 确认google_cloud_default连接配置：
  • 连接类型选择Google Cloud
  • Keyfile JSON字段粘贴完整的服务账号JSON密钥，无格式错误
  • 连接的项目ID与服务账号所属项目ID一致
• 确保Airflow的GCP Provider包（apache-airflow-providers-google）为最新版本，避免版本兼容问题
• 若在本地运行Airflow，检查是否设置了GOOGLE_APPLICATION_CREDENTIALS环境变量，该变量会干扰Airflow连接的凭证优先级

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