如何使用Python向Azure发起POST请求并解决认证失败问题
解决Azure Data Factory Airflow同步API的认证问题
Azure管理API不支持直接通过auth=(client_id, client_secret)这种基本认证方式,必须使用Azure AD OAuth2 Bearer令牌进行认证。以下是正确的实现步骤:
1. 获取Azure AD访问令牌
首先需要向Azure AD的令牌端点请求获取访问令牌,需要用到租户ID、客户端ID、客户端密钥:
import requests # 替换为你的实际信息 tenant_id = "your-tenant-id" client_id = "AZURE_CLIENT_ID" client_secret = "AZURE_CLIENT_SECRET" resource = "https://management.azure.com/" # 请求令牌 token_url = f"https://login.microsoftonline.com/{tenant_id}/oauth2/token" token_data = { "grant_type": "client_credentials", "client_id": client_id, "client_secret": client_secret, "resource": resource } token_response = requests.post(token_url, data=token_data) access_token = token_response.json().get("access_token") if not access_token: print("获取令牌失败:", token_response.json()) exit()
2. 携带令牌发起POST请求
拿到访问令牌后,在请求头中添加Authorization: Bearer {access_token},再调用目标API:
api_url = "https://management.azure.com/subscriptions/subscription_id/resourcegroups/resourcegroup_id/providers/Microsoft.DataFactory/factories/datafactory_id/airflow/sync?api-version=2018-06-01" todo = { "IntegrationRuntimeName": "Airflow", "LinkedServiceName": "LAIRFLOW", "StorageFolderPath": "airflow/dev/", "CopyFolderStructure": "true", "Overwrite": "true", } # 构造请求头,带上Bearer令牌 headers = { "Authorization": f"Bearer {access_token}", "Content-Type": "application/json" } response = requests.post(api_url, json=todo, headers=headers) print(response.json()) print(response.status_code)
关键说明
- 确保你的服务主体(客户端ID对应的应用)拥有Data Factory Contributor或足够的权限来执行Airflow同步操作
- 所有占位符(
tenant_id、subscription_id、resourcegroup_id等)都要替换为你的实际资源信息 - 令牌有有效期,过期后需要重新获取
内容的提问来源于stack exchange,提问作者Luiz
相关产品推荐
相关产品推荐

