从Databricks自动化更新Azure AD组成员遇错，求自动化方案

一、当前脚本错误修复 #

你遇到的400错误是因为传递了用户邮箱而非Azure AD用户的Object ID。Microsoft Graph API的groups/{group-id}/members/$ref接口要求@odata.id中使用用户的唯一Object ID，而非邮箱地址。

修正后的脚本逻辑 #

1. 从CSV读取用户邮箱列表
2. 通过Graph API根据邮箱查询对应的用户Object ID
3. 先清理Azure AD组内的现有成员（实现全量更新）
4. 批量添加新的用户成员

以下是修正后的Python脚本（适配Databricks环境）：

import requests
import pandas as pd

# 读取CSV文件（假设CSV路径为dbfs路径，列名为user_email）
df = pd.read_csv("/dbfs/mnt/reports/users.csv")
user_emails = df["user_email"].tolist()

# Azure AD与Graph API配置
tenant_id = "<tenant_id>"
client_id = dbutils.secrets.get(scope="<scope>", key="<client_id>")
client_secret = dbutils.secrets.get(scope="<scope>", key="<client_secret>")
group_id = "<group_id>"
graph_url = "https://graph.microsoft.com/v1.0"

# 获取访问令牌
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": "https://graph.microsoft.com"
}
token_response = requests.post(token_url, data=token_data)
access_token = token_response.json()["access_token"]
headers = {"Authorization": f"Bearer {access_token}", "Content-Type": "application/json"}

# 1. 查询组内现有成员并移除
def remove_all_group_members():
    members_url = f"{graph_url}/groups/{group_id}/members"
    while True:
        response = requests.get(members_url, headers=headers)
        if response.status_code != 200:
            print(f"获取成员失败: {response.text}")
            break
        members = response.json().get("value", [])
        if not members:
            break
        for member in members:
            remove_url = f"{graph_url}/groups/{group_id}/members/{member['id']}/$ref"
            remove_response = requests.delete(remove_url, headers=headers)
            if remove_response.status_code == 204:
                print(f"移除成员: {member['userPrincipalName']}")
        # 处理分页
        if "@odata.nextLink" in response.json():
            members_url = response.json()["@odata.nextLink"]
        else:
            break

# 2. 根据邮箱查询用户Object ID
def get_user_object_id(email):
    user_url = f"{graph_url}/users/{email}"
    response = requests.get(user_url, headers=headers)
    if response.status_code == 200:
        return response.json()["id"]
    else:
        print(f"查询用户失败 {email}: {response.text}")
        return None

# 3. 添加新成员到组
def add_user_to_group(user_id):
    add_url = f"{graph_url}/groups/{group_id}/members/$ref"
    payload = {"@odata.id": f"{graph_url}/directoryObjects/{user_id}"}
    response = requests.post(add_url, headers=headers, json=payload)
    if response.status_code == 204:
        print(f"添加成员成功: {user_id}")
    else:
        print(f"添加成员失败 {user_id}: {response.text}")

# 执行全量更新流程
remove_all_group_members()
for email in user_emails:
    user_id = get_user_object_id(email)
    if user_id:
        add_user_to_group(user_id)

二、自动化实现方案 #

方案1：Azure DevOps Pipeline 自动化 #

适合DevOps团队主导的流程，通过Pipeline每周触发执行更新：

• 步骤1：将修正后的脚本（PowerShell/Python）存入Azure DevOps Repo
• 步骤2：创建新Pipeline，配置每周定时触发器（比如每周一凌晨）
• 步骤3：添加任务：
  • 若用Python：安装依赖包requests、pandas，读取存储在Blob Storage或Repo中的CSV文件
  • 若用PowerShell：调用Graph API（可使用Microsoft.Graph模块简化操作）
• 步骤4：配置服务连接：使用Azure Resource Manager服务主体，授予其Group.ReadWrite.All和User.Read.All的应用权限（需在Azure AD门户中配置API权限并同意管理员授权）

方案2：Azure Data Factory (ADF) 自动化 #

适合数据团队使用现有ADF资源实现：

• 步骤1：将用户CSV文件存储在Azure Blob Storage中
• 步骤2：在ADF中创建管道：
  1. Lookup活动：读取Blob中的CSV文件，获取用户邮箱列表
  2. Databricks Notebook活动：调用修正后的Databricks脚本，传入邮箱列表参数，执行组清理与成员添加
  3. （可选）Web活动：直接调用Graph API完成操作（无需Databricks，适合轻量场景）
• 步骤3：配置触发器：设置每周定时触发管道
• 步骤4：权限配置：确保ADF使用的服务主体拥有Graph API的Group.ReadWrite.All和User.Read.All应用权限

关键权限说明 #

无论使用哪种方案，服务主体必须拥有以下Azure AD Graph API权限（应用权限类型，需管理员同意）：

• Group.ReadWrite.All：允许读写Azure AD组
• User.Read.All：允许查询用户信息

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