使用 Python 和 Microsoft Graph API 查询 Microsoft Entra 中的数据

微软的 Microsoft Graph API 可用于查询 Microsoft Entra 中的数据，比如企业中的用户目录等。同时，微软还提供了官方的 SDK ，以便开发者快速开发应用。在实际使用 SDK 时，尽管可以向 Copilot 提问，以获取使用方法。但在实际中，Copilot 所给出的使用方法可能过时或者不准确。

任务 🔗

列出当前组织 Microsoft Entra 里的所有 Service Principal。

技术选型 🔗

• 语言：Python
• SDK：Microsoft Graph SDK for Python

实现 🔗

参考 Microsoft Graph SDK for Python 项目的 README，可按部就班地编写程序完成任务。

安装 msgraph-sdk 🔗

pip install msgraph-sdk

需要注意的是，与之类似的 Python 包有 msgraph-core。 《Microsoft Graph Python SDK Upgrade Guide》 一文介绍了如何从msgraph-core 升级到 msgraph-sdk。

在简单的使用之后，本人的体会是：msgraph-sdk 似乎是 msgraph-core 的更高层的抽象和封装。除了接口本身不同，比较有意思的变化是，msgraph-sdk 似乎只提供了异步接口。这尽管提高了运行效率，但让初级的 Python 开发者理解理解和掌握异步的Python，还是需要一番功夫。

认证 🔗

大部分情况下，使用 DefaultAzureCredential 完成认证。它会尝试多种不同的认证机制，获取访问令牌。通常本地开发环境和生产环境是不同的，因此使用 DefaultAzureCredential 可实现一定程度的“自适应”。

from azure.identity import DefaultAzureCredential
credentials = DefaultAzureCredential()

查询数据 🔗

极简场景 🔗

在最简单的场景下，可使用如下方式获取数据：

import asyncio
from azure.identity import DefaultAzureCredential
from msgraph import GraphServiceClient

credentials = DefaultAzureCredential()

client = GraphServiceClient(credentials=credentials)

async def get_service_principals():
    response = await client.service_principals.get()
    return response.value

service_princials = asyncio.run(get_service_principals())

在以上代码中，get_service_principals() 函数实现了查询数据的功能。但是它至少有如下两个弊端：

1. 在使用 Graph API 查询数据时，默认只返回 100 条数据。如果组织里有超过 100 个 Service Principal，那么以上函数只能获取其中的 100 条记录。
2. 每一条 Service Principal 记录可能会比较大，但实际中可能只需要若个字段。

实战场景 🔗

《Paging Microsoft Graph data in your app》 一文介绍了 Microsoft Graph 的分页机制。简单来说，当查询列表数据时，服务器除了返回数据记录本身，还返回了 @odata.nextLink 属性，用于指示下一页的地址。

指定查询串中的参数，可自定义服务器返回的数据，见《Customize Microsoft Graph responses with query parameters 》(https://learn.microsoft.com/en-us/graph/query-parameters) 一文。如果指向返回指定的字段，那么可使用 $select 来指定。

实现如下：

import asyncio
from azure.identity import DefaultAzureCredential

from msgraph import GraphServiceClient
from kiota_abstractions.base_request_configuration import RequestConfiguration
from msgraph.generated.service_principals.service_principals_request_builder import ServicePrincipalsRequestBuilder

credentials = DefaultAzureCredential()
client = GraphServiceClient(credentials=credentials)

async def get_service_principals():
    service_princials = []
    odata_next_link = None

    while True:
        if not odata_next_link:
            query_params = ServicePrincipalsRequestBuilder.ServicePrincipalsRequestBuilderGetQueryParameters(
                select = ["displayName"],
            )

            request_configuration = RequestConfiguration(query_parameters=query_params)
            request_configuration.headers.add("ConsistencyLevel", "eventual")

            response = await client.service_principals.get(request_configuration = request_configuration)
        else:
            response = await client.service_principals.with_url(odata_next_link).get()

        service_princials.extend(response.value)

        if not response.odata_next_link:
            break

        odata_next_link = response.odata_next_link

    return service_princials

service_princials = asyncio.run(get_service_principals())