通过 Python 开始使用适用于 NoSQL 的 Azure Cosmos DB

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

包 （PyPi） | API 参考 | 库源代码 | 提供反馈

先决条件

• 具有活动订阅的 Azure 帐户。 创建试用版订阅。
• Azure Cosmos DB for NoSQL 帐户。 了解如何 为 NoSQL 帐户创建 API。
• Python 3.7 或更高版本。
• Azure Command-Line 接口（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 并创建它的实例。

• 使用终结点和凭据作为参数创建 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 参考
• 库源代码
• 提供反馈

后续步骤