Azure Cosmos DB Client Library for Python

4.15.0 · active · verified Wed Apr 01

The `azure-cosmos` library is the official Microsoft Azure Cosmos DB client library for Python. It provides a client-side logical representation to access Azure Cosmos DB for NoSQL accounts, enabling developers to manage databases, containers (collections of JSON documents), and items (individual JSON documents). It supports common operations like CRUD (Create, Read, Update, Delete) and SQL-like querying. The library is actively maintained, with the current stable version being 4.15.0, and receives regular updates and feature enhancements.

Warnings

Install

Imports

Quickstart

This quickstart demonstrates how to connect to Azure Cosmos DB, create a database and container (if they don't exist), perform CRUD operations (Create, Read, Update, Delete) on items, and query data using the `azure-cosmos` SDK. It retrieves Cosmos DB account URI and key from environment variables, defaulting to Azure Cosmos DB Emulator settings if not found. Make sure to have a Cosmos DB account (or emulator) running and environment variables `ACCOUNT_URI` and `ACCOUNT_KEY` configured.

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

# These environment variables must be set for the quickstart to run
# In Azure Portal, navigate to your Cosmos DB account -> Keys
# ACCOUNT_URI = "<your-cosmos-db-endpoint>"
# ACCOUNT_KEY = "<your-cosmos-db-primary-key>"

ACCOUNT_URI = os.environ.get('ACCOUNT_URI', 'https://localhost:8081') # Use emulator default if not set
ACCOUNT_KEY = os.environ.get('ACCOUNT_KEY', 'C2y6yDjf5/R+ob0N8A7Cgv30VRDJIWEHLM+4QDU5DE2nQ9ndKll+E/6DrURj/Dd/C04hZlwsEaLicensedy6yDjf5/R+ob0N8A7Cgv30VRDJIWEHLM+4QDU5DE2nQ9ndKll+E/6DrURj/Dd/C04hZlwsEaLicensed==') # Use emulator default if not set

DATABASE_NAME = 'myDatabase'
CONTAINER_NAME = 'myContainer'
PARTITION_KEY_PATH = '/category'

def main():
    if not ACCOUNT_URI or not ACCOUNT_KEY:
        print("Please set the ACCOUNT_URI and ACCOUNT_KEY environment variables.")
        print("You can find these in your Azure Cosmos DB account's 'Keys' section.")
        print("For local emulator, ensure it's running and use default key.")
        return

    try:
        # Initialize the Cosmos client
        client = CosmosClient(ACCOUNT_URI, credential=ACCOUNT_KEY)
        print("CosmosClient initialized.")

        # Create a database if it doesn't exist
        try:
            database = client.create_database_if_not_exists(id=DATABASE_NAME)
            print(f"Database '{DATABASE_NAME}' created or already exists.")
        except exceptions.CosmosHttpResponseError as e:
            print(f"Error creating/getting database: {e}")
            return

        # Create a container if it doesn't exist
        try:
            container = database.create_container_if_not_exists(
                id=CONTAINER_NAME,
                partition_key=PartitionKey(path=PARTITION_KEY_PATH),
                offer_throughput=400 # Or specify autoscale settings
            )
            print(f"Container '{CONTAINER_NAME}' created or already exists with partition key '{PARTITION_KEY_PATH}'.")
        except exceptions.CosmosHttpResponseError as e:
            print(f"Error creating/getting container: {e}")
            return

        # Create an item
        new_item = {
            'id': 'item1',
            'name': 'Laptop',
            'category': 'Electronics',
            'price': 1200.00
        }
        try:
            created_item = container.upsert_item(body=new_item)
            print(f"Item created/updated: {created_item['id']}")
        except exceptions.CosmosHttpResponseError as e:
            print(f"Error creating/updating item: {e}")
            return

        # Read an item
        try:
            read_item = container.read_item(item=new_item['id'], partition_key=new_item['category'])
            print(f"Item read: {read_item['name']}, Price: {read_item['price']}")
        except exceptions.CosmosHttpResponseError as e:
            print(f"Error reading item: {e}")
            return

        # Query items
        query = f"SELECT * FROM c WHERE c.category = '{new_item['category']}'"
        items = list(container.query_items(query=query, enable_cross_partition_query=False))
        print(f"Items found by query:")
        for item in items:
            print(f"  - {item['name']}")

        # Delete an item
        try:
            container.delete_item(item=new_item['id'], partition_key=new_item['category'])
            print(f"Item '{new_item['id']}' deleted.")
        except exceptions.CosmosHttpResponseError as e:
            print(f"Error deleting item: {e}")
            return

    except Exception as ex:
        print(f"An unexpected error occurred: {ex}")

if __name__ == '__main__':
    main()

view raw JSON →