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

适用范围： NoSQL

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

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

先决条件

• 具有活动订阅的 Azure 帐户。 创建试用版订阅。
• Azure Cosmos DB for NoSQL 帐户。 创建 API for NoSQL 帐户。
• 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 并创建它的实例。

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

后续步骤