使用 Python 在 Azure Cosmos DB for NoSQL 中创建容器

适用范围： NoSQL

Azure Cosmos DB 中的容器存储项集。 必须先创建容器，然后才能创建、查询或管理项。

为容器命名

在 Azure Cosmos DB 中，容器类似于关系数据库中的表。 创建容器时，容器名称构成用于访问容器资源和任何子项的 URI 段。

创建后，容器的 URI 采用以下格式：

https://<cosmos-account-name>.documents.azure.cn/dbs/<database-name>/colls/<container-name>

创建容器

若要创建容器，请调用以下方法之一：

• CreateContainerAsync
• CreateContainerIfNotExistsAsync
• 异步创建容器

创建容器

以下示例使用 DatabaseProxy.create_container 方法创建一个容器。 如果已存在同名容器，此方法会引发异常。

"""Sample showing how to connect with endpoint and key."""
import os

from azure.cosmos import CosmosClient, PartitionKey
from azure.cosmos.exceptions import (
    CosmosHttpResponseError,
    CosmosResourceExistsError,
)

DATABASE_ID = "cosmicworks-1"
CONTAINER_ID = "products"
ENDPOINT = os.environ["COSMOS_ENDPOINT"]
KEY = os.environ["COSMOS_KEY"]

client = CosmosClient(url=ENDPOINT, credential=KEY)


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

        # <create_container>
        try:
            partition_key_path = PartitionKey(path="/categoryId")
            container = database.create_container(
                id=CONTAINER_ID,
                partition_key=partition_key_path,
                offer_throughput=400,
            )
            print(f"Container created: {container.id}")

        except CosmosResourceExistsError:
            print("Container already exists.")
        # </create_container>

    except CosmosHttpResponseError:
        print("Request to the Azure Cosmos database service failed.")


if __name__ == "__main__":
    main()

创建一个容器（如果不存在）

以下示例使用 DatabaseProxy.create_container_if_not_exists 方法创建一个容器。 与以前的 create 方法相比，如果数据库已存在，此方法不会引发异常。 如果多次运行同一代码，则此方法对于避免错误非常有用。

"""Sample showing how to connect with endpoint and key."""
import os

from azure.cosmos import CosmosClient, PartitionKey
from azure.cosmos.exceptions import CosmosHttpResponseError

DATABASE_ID = "cosmicworks-3"
CONTAINER_ID = "products"
ENDPOINT = os.environ["COSMOS_ENDPOINT"]
KEY = os.environ["COSMOS_KEY"]

client = CosmosClient(url=ENDPOINT, credential=KEY)


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

        # <create_container>
        try:
            partition_key_path = PartitionKey(path="/categoryId")
            container = database.create_container_if_not_exists(
                id=CONTAINER_ID,
                partition_key=partition_key_path,
                offer_throughput=400,
            )
            print(f"Container created or returned: {container.id}")

        except CosmosHttpResponseError:
            print("Request to the Azure Cosmos database service failed.")
        # </create_container>

        try:
            # <parse_response>
            partition_key_path = PartitionKey(path="/categoryId")
            container = database.create_container_if_not_exists(
                id=CONTAINER_ID,
                partition_key=partition_key_path,
                offer_throughput=400,
            )
            for doc in container.read_all_items(max_item_count=10):
                print(f'Doc id: {doc["id"]}')
            # </parse_response>

        except CosmosHttpResponseError:
            print("Request to the Azure Cosmos database service failed.")

    except CosmosHttpResponseError:
        print("Request to the Azure Cosmos database service failed.")


if __name__ == "__main__":
    main()

异步创建容器

还可以使用 azure.cosmos.aio 命名空间中的类似对象和方法异步创建数据库。 例如，使用 DatabaseProxy.create_database 方法或 CosmoClient.create_database_if_not_exists 方法。

当你想要并行执行多个操作时，以异步方式工作非常有用。 有关详细信息，请参阅使用异步客户端。

分析响应

在上面的示例中，来自请求的响应是 ContainerProxy，是一个与 DB 容器交互的接口。 在代理中，可以访问那些用于对容器执行操作的方法。

以下示例演示返回 container 对象的 create_container_if_not_exists 方法。

"""Sample showing how to connect with endpoint and key."""
import os

from azure.cosmos import CosmosClient, PartitionKey
from azure.cosmos.exceptions import CosmosHttpResponseError

DATABASE_ID = "cosmicworks-3"
CONTAINER_ID = "products"
ENDPOINT = os.environ["COSMOS_ENDPOINT"]
KEY = os.environ["COSMOS_KEY"]

client = CosmosClient(url=ENDPOINT, credential=KEY)


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

        # <create_container>
        try:
            partition_key_path = PartitionKey(path="/categoryId")
            container = database.create_container_if_not_exists(
                id=CONTAINER_ID,
                partition_key=partition_key_path,
                offer_throughput=400,
            )
            print(f"Container created or returned: {container.id}")

        except CosmosHttpResponseError:
            print("Request to the Azure Cosmos database service failed.")
        # </create_container>

        try:
            # <parse_response>
            partition_key_path = PartitionKey(path="/categoryId")
            container = database.create_container_if_not_exists(
                id=CONTAINER_ID,
                partition_key=partition_key_path,
                offer_throughput=400,
            )
            for doc in container.read_all_items(max_item_count=10):
                print(f'Doc id: {doc["id"]}')
            # </parse_response>

        except CosmosHttpResponseError:
            print("Request to the Azure Cosmos database service failed.")

    except CosmosHttpResponseError:
        print("Request to the Azure Cosmos database service failed.")


if __name__ == "__main__":
    main()

后续步骤

创建容器后，请使用下一个指南来创建项。