使用 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() 函数实现了查询数据的功能。但是它至少有如下两个弊端：

在使用 Graph API 查询数据时，默认只返回 100 条数据。如果组织里有超过 100 个 Service Principal，那么以上函数只能获取其中的 100 条记录。
每一条 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())