如何配置带MFA（应用+用户）的Partner Center API并通过Python/Spark调用

问题原因 #

你当前使用的client_credentials（仅应用）认证方式仅适用于应用权限，而部分Partner Center API接口要求委托权限（需要用户上下文），因此必须切换到支持用户身份的认证流程（如设备码流，适合无UI的Spark环境）。

前置配置 #

• 在Azure AD应用注册中添加委托权限：搜索并勾选Access Partner Center API，完成后授予管理员同意
• 确保应用注册的"认证"页面中，启用"设备码流"（在"高级设置"的"允许公共客户端流"中开启）

修改后的代码 #

import requests
import time
from pyspark.sql import SparkSession
from pyspark.sql.types import StructField, StringType, StructType

# 认证配置
auth_config = {
    "tenant_id": "<你的租户ID>",
    "client_id": mssparkutils.credentials.getSecret('https://vaultname.vault.azure.net/', 'KV-Client-ID'),
    "resource": "https://api.partnercenter.microsoft.com",  # Partner Center API的正确资源标识
    "auth_url": "https://login.microsoftonline.com/{tenant_id}/oauth2/v2.0/devicecode",
    "token_url": "https://login.microsoftonline.com/{tenant_id}/oauth2/v2.0/token"
}

# 定义DataFrame Schema
_schema = StructType([
    StructField("Tenant ID", StringType(), True),
    StructField("Domain", StringType(), True),
    StructField("Company", StringType(), True),
    StructField("URI", StringType(), True)
])

def get_access_token():
    # 第一步：获取设备码
    device_code_url = auth_config["auth_url"].format(tenant_id=auth_config["tenant_id"])
    device_code_payload = {
        "client_id": auth_config["client_id"],
        "scope": f"{auth_config['resource']}/user_impersonation offline_access"  # offline_access用于获取刷新令牌
    }
    device_code_response = requests.post(device_code_url, data=device_code_payload)
    device_code_response.raise_for_status()
    device_code_data = device_code_response.json()
    
    # 输出设备码信息，提示用户登录
    print(f"请访问: {device_code_data['verification_uri']}")
    print(f"输入代码: {device_code_data['user_code']}")
    print("等待用户登录...")
    
    # 第二步：轮询获取令牌
    token_payload = {
        "client_id": auth_config["client_id"],
        "grant_type": "urn:ietf:params:oauth:grant-type:device_code",
        "device_code": device_code_data["device_code"]
    }
    token_url = auth_config["token_url"].format(tenant_id=auth_config["tenant_id"])
    
    while True:
        token_response = requests.post(token_url, data=token_payload)
        token_data = token_response.json()
        
        if token_response.status_code == 200:
            # 返回访问令牌和刷新令牌（可选，用于后续无需重新登录）
            return token_data["access_token"], token_data.get("refresh_token")
        elif token_data.get("error") == "authorization_pending":
            # 等待后重试
            time.sleep(device_code_data["interval"])
        else:
            raise Exception(f"认证失败: {token_data.get('error_description')}")

def fetch_partner_center_data(access_token):
    api_url = "https://api.partnercenter.microsoft.com/v1/customers"
    headers = {
        "Authorization": f"Bearer {access_token}",
        "Accept": "application/json",
        "MS-PartnerCenter-Client": "Python-Spark"  # 标识客户端，可选
    }
    response = requests.get(api_url, headers=headers)
    response.raise_for_status()
    return response.json()

def create_spark_df(spark, results):
    return spark.createDataFrame(results, schema=_schema)

if __name__ == "__main__":
    spark = SparkSession.builder.appName("PartnerCenter-API-Call").getOrCreate()
    all_results = []
    
    # 获取令牌
    access_token, refresh_token = get_access_token()
    
    # 获取API数据
    api_data = fetch_partner_center_data(access_token)
    
    # 解析数据
    for item in api_data["items"]:
        profile = item["companyProfile"]
        tenant_id = profile.get("tenantId", "N/A").strip()
        domain = profile.get("domain", "N/A").strip()
        company_name = profile.get("companyName", "N/A").strip()
        uri = item["links"]["self"]["uri"]
        
        all_results.append((tenant_id, domain, company_name, uri))
    
    # 创建并显示DataFrame
    df = create_spark_df(spark, all_results)
    df.show(truncate=False)

关键说明 #

1. 设备码流优势：适合Spark这类无UI的服务器环境，通过用户在其他设备上登录完成认证，无需前端页面
2. 资源标识修正：将原代码中的https://graph.windows.net改为Partner Center API的正确资源https://api.partnercenter.microsoft.com
3. 离线访问权限：添加offline_access scope可获取刷新令牌，后续无需重复用户登录，直接用刷新令牌获取新的访问令牌
4. 错误处理：添加了raise_for_status()确保请求失败时抛出异常，便于调试

刷新令牌复用（可选） #

如果需要长期运行，可将刷新令牌存储在密钥保管库中，下次直接用刷新令牌获取访问令牌：

def refresh_access_token(refresh_token):
    token_payload = {
        "client_id": auth_config["client_id"],
        "grant_type": "refresh_token",
        "refresh_token": refresh_token,
        "scope": f"{auth_config['resource']}/user_impersonation offline_access"
    }
    token_url = auth_config["token_url"].format(tenant_id=auth_config["tenant_id"])
    response = requests.post(token_url, data=token_payload)
    response.raise_for_status()
    token_data = response.json()
    return token_data["access_token"], token_data["refresh_token"]

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