Manage Azure Cosmos DB Containers with Python SDK

This guide details how to interact with and manage containers within your Azure Cosmos DB database using the official Python SDK. You'll learn to create, read, update, and delete containers, as well as understand key container properties.

Before you begin, ensure you have:

• An Azure subscription.
• An Azure Cosmos DB account set up.
• The Azure Cosmos DB Python SDK installed: pip install azure-cosmos
• Your Cosmos DB account endpoint and key (available in the Azure portal).

First, establish a connection to your Cosmos DB account. You'll need your endpoint and primary key.

from azure.cosmos import CosmosClient

# Replace with your actual endpoint and key
endpoint = "YOUR_COSMOS_DB_ENDPOINT"
key = "YOUR_COSMOS_DB_PRIMARY_KEY"

client = CosmosClient(endpoint, credential=key)
        

Once connected, you can access your databases and then specifically target containers within those databases.

# Replace 'YourDatabaseName' with the name of your database
database_name = "YourDatabaseName"
database = client.get_database_client(database_name)

# Replace 'YourContainerName' with the name of your container
container_name = "YourContainerName"
container = database.get_container_client(container_name)
        

To create a new container, you'll specify its ID and a partition key path. The partition key is crucial for distributing data across partitions and ensuring scalability.

container_id = "myNewContainer"
partition_key_path = "/myPartitionKey" # Example: "/category" or "/userId"

try:
    container_definition = {
        "id": container_id,
        "partitionKey": {"paths": [partition_key_path]}
    }
    created_container = database.create_container(body=container_definition)
    print(f"Container '{created_container['id']}' created successfully.")
except Exception as e:
    print(f"Error creating container: {e}")
        

You can retrieve information about an existing container, such as its ID, throughput settings, and partition key definition.

container_to_read_id = "myNewContainer"
container_read_client = database.get_container_client(container_to_read_id)

container_properties = container_read_client.read()

print(f"Container ID: {container_properties['id']}")
print(f"Partition Key Path: {container_properties['partitionKey']['paths'][0]}")
print(f"Self Link: {container_properties['_self']}")
# print(f"Throughput (RU/s): {container_properties['throughput']}") # Note: Throughput might require a separate query or specific SDK call if not directly in read()
        

While the partition key cannot be changed after creation, you can adjust other settings like indexing policies or TTL (Time-to-Live).

Note: Directly modifying the partition key path is not supported. If you need to change the partition key, you must create a new container with the desired key and migrate your data.

container_to_update_id = "myNewContainer"
container_update_client = database.get_container_client(container_to_update_id)

# Example: Updating indexing policy (simplified)
# For detailed indexing policy configuration, refer to Azure Cosmos DB documentation.
indexing_policy = {
    "indexingMode": "consistent",
    "automatic": True,
    "includedPaths": [
        {"path": "/*"}
    ],
    "excludedPaths": [
        {"path": "/\"_etag\"/?\n"}
    ]
}

updated_properties = container_update_client.replace_container(
    indexing_policy=indexing_policy
    # You can also update ttl here if applicable
)

print(f"Container '{updated_properties['id']}' updated successfully.")
        

Removing a container is a permanent action and will delete all data within it. Use with caution.

container_to_delete_id = "myNewContainer"
container_delete_client = database.get_container_client(container_to_delete_id)

try:
    container_delete_client.delete_container()
    print(f"Container '{container_to_delete_id}' deleted successfully.")
except Exception as e:
    print(f"Error deleting container: {e}")
        

Retrieve a list of all containers within a specific database.

print(f"Containers in database '{database_name}':")
for container_item in database.list_containers():
    print(f"- {container_item['id']}")
        

Throughput is provisioned in Request Units per second (RU/s). You can scale this for your containers.

Reading Container Throughput

container_throughput_id = "MyContainerWithThroughput" # Assumes this container exists
container_throughput_client = database.get_container_client(container_throughput_id)

# The SDK might require a specific method or query to get RU/s directly.
# Often, you can infer it from container properties or use a separate operation.
# For explicit RU/s, you might need to query offers or use the `read_throughput` method.
# Example using `read_throughput` (check SDK version for exact availability):
try:
    throughput_info = container_throughput_client.read_throughput()
    print(f"Current throughput for '{container_throughput_id}': {throughput_info.content['throughput']} RU/s")
except Exception as e:
    print(f"Could not read throughput directly: {e}")
    print("Throughput information might be available in container properties or requires a different API call.")

        

Requesting Container Throughput (Scaling)

To change the RU/s, you'll typically use the replace_container_throughput method or adjust offers.

container_to_scale_id = "MyContainerWithThroughput"
new_throughput = 600 # Example: scale to 600 RU/s

container_scale_client = database.get_container_client(container_to_scale_id)

try:
    updated_throughput_info = container_scale_client.replace_container_throughput(throughput=new_throughput)
    print(f"Throughput for '{container_to_scale_id}' updated to {updated_throughput_info.content['throughput']} RU/s.")
except Exception as e:
    print(f"Error updating throughput: {e}")
        

Autoscale: For automatic scaling, configure autoscale settings via the Azure portal or specific SDK methods related to offer types.

• Partition Key Choice: Selecting an effective partition key is critical for performance and scalability. Choose a key with high cardinality and even distribution.
• Indexing Policy: Optimize your indexing policy to improve query performance while managing storage and RU costs.
• Throughput Management: Monitor your RU consumption and adjust throughput accordingly to balance performance and cost. Use autoscale where appropriate.
• Error Handling: Always implement robust error handling, especially for operations like creation and deletion.
• SDK Versions: Ensure you are using a recent version of the azure-cosmos SDK for the latest features and bug fixes.