次の方法で共有

使用 Azure DocumentDB 生成 Python 控制台应用

本指南逐步讲解如何构建 Python 控制台应用程序以连接到 Azure DocumentDB 群集。 配置开发环境，使用 azure.identity 用于 Python 的 Azure SDK 中的包进行身份验证，并执行创建、查询和更新文档等作。

先决条件

  • Azure 订阅服务

    • 如果没有 Azure 订阅，请创建 试用版

  • 现有的 Azure DocumentDB 群集

  • 如需在本地运行 CLI 参考命令，请安装 Azure CLI。 如果在 Windows 或 macOS 上运行，请考虑在 Docker 容器中运行 Azure CLI。 有关详细信息，请参阅如何在 Docker 容器中运行 Azure CLI。

    • 如果使用的是本地安装，请使用 az login 命令登录到 Azure CLI。 若要完成身份验证过程，请遵循终端中显示的步骤。 有关其他登录选项，请参阅使用 Azure CLI 登录。

    • 出现提示时，请在首次使用时安装 Azure CLI 扩展。 有关扩展的详细信息，请参阅 将扩展与 Azure CLI 配合使用。

    • 运行az version命令，以查看已安装的版本和依赖库。 若要升级到最新版本，请运行az upgrade。

  • 为群集配置了 Microsoft Entra 身份验证，已授予标识 root 角色。

  • 最新版本的 Python。

配置控制台应用程序

接下来，创建新的控制台应用程序项目，并导入必要的库以向群集进行身份验证。

  1. 为项目创建新目录并设置虚拟环境。

    mkdir mongodb-app
cd mongodb-app
python -m venv .venv

  2. 激活虚拟环境。

    # On Windows
.venv\Scripts\activate

# On macOS/Linux
source .venv/bin/activate

  3. 为应用程序创建新的 Python 文件。

    touch app.py

  4. 安装用于 Azure 身份验证的 azure.identity 库。

    pip install azure.identity

  5. 安装适用于 Python 的 pymongo 驱动程序。

    pip install pymongo

连接至群集

现在，使用 Azure.Identity 库获取一个 TokenCredential，以用于连接到您的群集。 官方 MongoDB 驱动程序具有一个特殊接口，必须实现该接口，以便从 Microsoft Entra 获取令牌，以便在连接到群集时使用。

  1. 在 Python 文件顶部导入必要的模块。

    from azure.identity import DefaultAzureCredential
from pymongo import MongoClient
from pymongo.auth_oidc import OIDCCallback, OIDCCallbackContext, OIDCCallbackResult

  2. 创建实现 MongoDB OpenID Connect （OIDC） 回调接口的自定义类。

    class AzureIdentityTokenCallback(OIDCCallback):
    def __init__(self, credential):
        self.credential = credential

    def fetch(self, context: OIDCCallbackContext) -> OIDCCallbackResult:
        token = self.credential.get_token(
            "https://ossrdbms-aad.database.chinacloudapi.cn/.default").token
        return OIDCCallbackResult(access_token=token)

  3. 设置群集名称变量。

    clusterName = "<azure-cosmos-db-mongodb-vcore-cluster-name>"

  4. 创建 DefaultAzureCredential 的实例并设置身份验证属性。

    credential = DefaultAzureCredential()
authProperties = {"OIDC_CALLBACK": AzureIdentityTokenCallback(credential)}

  5. 创建配置了 Microsoft Entra 身份验证的 MongoDB 客户端。

    client = MongoClient(
    f"mongodb+srv://{clusterName}.global.mongocluster.cosmos.azure.cn/",
    connectTimeoutMS=120000,
    tls=True,
    retryWrites=True,
    authMechanism="MONGODB-OIDC",
    authMechanismProperties=authProperties
)

print("Client created")

执行常见操作

最后，使用官方库对数据库、集合和文档执行常见任务。 在这里，你将使用相同的类和方法与 MongoDB 或 DocumentDB 进行交互来管理集合和项。

  1. 获取对数据库的引用。

    database = client.get_database("<database-name>")

print("Database pointer created")

  2. 获取对集合的引用。

    collection = database.get_collection("<container-name>")

print("Collection pointer created")

  3. 创建文档并使用 将其更新插入到集合中。collection.update_one

    new_document = {
    "_id": "aaaaaaaa-0000-1111-2222-bbbbbbbbbbbb",
    "category": "gear-surf-surfboards",
    "name": "Yamba Surfboard",
    "quantity": 12,
    "price": 850.00,
    "clearance": False,
}

filter = {
    "_id": "aaaaaaaa-0000-1111-2222-bbbbbbbbbbbb",
}
payload = {
    "$set": new_document
}
result = collection.update_one(filter, payload, upsert=True)

  4. 用于 collection.find_one 从集合中检索特定文档。

    filter = {
    "_id": "aaaaaaaa-0000-1111-2222-bbbbbbbbbbbb",
    "category": "gear-surf-surfboards"
}
existing_document = collection.find_one(filter)
print(f"Read document _id:\t{existing_document['_id']}")

  5. 查询与 collection.find 筛选器匹配的多个文档。

    filter = {
    "category": "gear-surf-surfboards"
}
matched_documents = collection.find(filter)

for document in matched_documents:
    print(f"Found document:\t{document}")

相关内容

  • Last updated on