azure_cosmos_db_nosql_database_container_crud_and_query.py

1import azure.cosmos.documents as documents
2import azure.cosmos.cosmos_client as cosmos_client
3import azure.cosmos.exceptions as exceptions
4from azure.cosmos.partition_key import PartitionKey
5
6# Replace with your own configuration
7HOST = 'https://YOUR_ACCOUNT_NAME.documents.azure.com:443/'
8MASTER_KEY = 'YOUR_ACCOUNT_KEY'
9DATABASE_ID = 'SampleDatabase'
10CONTAINER_ID = 'SampleContainer'
11
12def run_sample():
13    client = cosmos_client.CosmosClient(HOST, {'masterKey': MASTER_KEY})
14
15    try:
16        # Create a database
17        try:
18            db = client.create_database(id=DATABASE_ID)
19            print(f"Database created: {db.id}")
20        except exceptions.CosmosResourceExistsError:
21            db = client.get_database_client(DATABASE_ID)
22            print(f"Database reached: {db.id}")
23
24        # Create a container
25        try:
26            container = db.create_container(
27                id=CONTAINER_ID, 
28                partition_key=PartitionKey(path='/partitionKey')
29            )
30            print(f"Container created: {container.id}")
31        except exceptions.CosmosResourceExistsError:
32            container = db.get_container_client(CONTAINER_ID)
33            print(f"Container reached: {container.id}")
34
35        # Create an item
36        item_data = {
37            'id': 'item1',
38            'partitionKey': 'pk1',
39            'name': 'Sample Item',
40            'description': 'This is a test item'
41        }
42        container.upsert_item(body=item_data)
43        print("Item upserted")
44
45        # Read item
46        response = container.read_item(item='item1', partition_key='pk1')
47        print(f"Item read: {response['name']}")
48
49        # Query items
50        items = list(container.query_items(
51            query="SELECT * FROM r WHERE r.partitionKey=@pk",
52            parameters=[{"name": "@pk", "value": "pk1"}],
53            enable_cross_partition_query=False
54        ))
55        print(f"Query returned {len(items)} items")
56
57    except exceptions.CosmosHttpResponseError as e:
58        print(f"An error occurred: {e.message}")
59
60if __name__ == '__main__':
61    run_sample()