通过 Python 开始使用 Azure Cosmos DB for NoSQL

本文介绍如何使用 Python SDK 连接到 Azure Cosmos DB for NoSQL。 连接后，对数据库、容器和项执行操作。

包 (PyPi) | API 参考 | 库源代码 | 给予反馈

先决条件

• 拥有有效订阅的 Azure 帐户。 创建试用版订阅。
• Azure Cosmos DB for NoSQL 帐户。 了解如何为 NoSQL 帐户创建 API。
• Python 3.7 或更高版本。
• Azure 命令行接口 (CLI) 或 Azure PowerShell。

设置你的项目

为 Python 代码创建环境。

pip install azure-cosmos

pip install azure-cosmos

创建 Python 应用程序

在你的环境中，创建新的 app.py 文件，并添加以下代码：

"""Sample showing how to connect with endpoint and key."""
# <imports>
import json
import os
import sys
import uuid

from azure.core.exceptions import AzureError
from azure.cosmos import CosmosClient, PartitionKey

# </imports>

DATABASE_ID = "cosmicworks"
CONTAINER_ID = "products"
# <client>
ENDPOINT = os.environ["COSMOS_ENDPOINT"]
KEY = os.environ["COSMOS_KEY"]

client = CosmosClient(url=ENDPOINT, credential=KEY)
# </client>


def main():
    """How to CosmosDB and NoSQL samples."""
    try:
        # Create database and partition key.
        database = client.create_database_if_not_exists(id=DATABASE_ID)

        # Create a container.
        partition_key_path = PartitionKey(path="/categoryId")
        container = database.create_container_if_not_exists(
            id=CONTAINER_ID,
            partition_key=partition_key_path,
            offer_throughput=400,
        )

        # Create a new item.
        new_guid = str(uuid.uuid4())
        new_item = {
            "id": new_guid,
            "categoryId": "61dba35b-4f02-45c5-b648-c6badc0cbd79",
            "categoryName": "gear-surf-surfboards",
            "name": "Yamba Surfboard",
            "quantity": 12,
            "sale": False,
        }
        container.create_item(new_item)

        # Query items.
        sql_stmt = "SELECT * FROM products p WHERE p.categoryId = @categoryId"
        category_id = "61dba35b-4f02-45c5-b648-c6badc0cbd79"
        params = [dict(name="@categoryId", value=category_id)]

        items = container.query_items(
            query=sql_stmt,
            parameters=params,
            enable_cross_partition_query=False,
        )

        for item in items:
            print(json.dumps(item, indent=True))

    except AzureError as err:
        sys.exit("Error:" + str(err))


if __name__ == "__main__":
    main()

以上代码将导入你要在本文余下内容中使用的模块。

连接到 Azure Cosmos DB for NoSQL

若要连接到 Azure Cosmos DB API for NoSQL，请创建 CosmosClient 类的实例。 此类是针对数据库执行所有操作的起点。

若要使用 Microsoft Entra 连接到 NoSQL 帐户的 API，请使用安全主体。 主体的确切类型将取决于应用程序代码的托管位置。 下表是快速参考指南。

导入 Azure.Identity

azure-identity 包提供在所有 Azure SDK 库之间共享的核心身份验证功能。

将 azure-identity 包导入环境。

pip install azure-identity

使用默认凭据实现创建 CosmosClient

如果在本地计算机上进行测试，或者应用程序在 Azure 服务上运行，并直接支持托管标识，则通过创建 DefaultAzureCredential 实例来获取 OAuth 令牌。

在 app.py 文件中：

• 获取帐户的终结点，并将其设置为环境变量 COSMOS_ENDPOINT。

• 导入 DefaultAzureCredential 并创建它的实例。

• 使用 ENDPOINT 和 credential 作为参数创建 CosmosClient 类的新实例。

"""Sample showing how to connect with AAD."""
import json
import os
import sys
import uuid

from azure.core.exceptions import AzureError
from azure.cosmos import CosmosClient

# <credential>
from azure.identity import DefaultAzureCredential

ENDPOINT = os.environ["COSMOS_ENDPOINT"]

credential = DefaultAzureCredential()

client = CosmosClient(ENDPOINT, credential)
# </credential>

DATABASE_ID = "cosmicworks"
CONTAINER_ID = "products"


def main():
    """How to CosmosDB and NoSQL samples."""
    try:
        # Get database.
        database = client.get_database_client(DATABASE_ID)

        # Get container.
        container = database.get_container_client(CONTAINER_ID)
        print("Container info: " + str(container.read()))

        # Create a new item.
        new_guid = str(uuid.uuid4())
        new_item = {
            "id": new_guid,
            "categoryId": "61dba35b-4f02-45c5-b648-c6badc0cbd79",
            "categoryName": "gear-surf-surfboards",
            "name": "Yamba Surfboard",
            "quantity": 12,
            "sale": False,
        }
        container.create_item(new_item)

        # Query items.
        sql_stmt = "SELECT * FROM products p WHERE p.categoryId = @categoryId"
        category_id = "61dba35b-4f02-45c5-b648-c6badc0cbd79"
        params = [dict(name="@categoryId", value=category_id)]

        items = container.query_items(
            query=sql_stmt,
            parameters=params,
            enable_cross_partition_query=False,
        )

        for item in items:
            print(json.dumps(item, indent=True))

    except AzureError as err:
        sys.exit("Error:" + str(err))


if __name__ == "__main__":
    main()

生成应用程序

在生成应用程序时，代码主要与四种类型的资源交互：

• API for NoSQL 帐户，它是 Azure Cosmos DB 数据的唯一顶级命名空间。

• 数据库，它用于组织帐户中的容器。

• 容器，它包含数据库中的一组单个项。

• 项，它表示容器中的 JSON 文档。

此图示显示了这些资源之间的关系。

一个或多个关联的 Python 类表示每种类型的资源。 此列表显示最常用的同步编程类。 （azure.cosmos.aio 命名空间中有类似的异步编程类。）

这些指南介绍了如何使用上述每个类来生成应用程序。

另请参阅

• PyPi
• API 参考
• 库源代码
• 提供反馈

后续步骤