Scribd
Upload
11 views

CRUD Operations with Azure Cosmos DB in Python

The document provides a comprehensive guide on performing CRUD operations with Azure Cosmos DB using Python. It details how to initialize the Cosmos client, create, read, update, and delete items, along with error handling for potential issues. Each operation is accompanied by code snippets to demonstrate the implementation process.

Uploaded by

Shabir Ahamed
Copyright
© © All Rights Reserved
We take content rights seriously. If you suspect this is your content, claim it here.
Available Formats
Download as DOCX, PDF, TXT or read online on Scribd
11 views

CRUD Operations with Azure Cosmos DB in Python

The document provides a comprehensive guide on performing CRUD operations with Azure Cosmos DB using Python. It details how to initialize the Cosmos client, create, read, update, and delete items, along with error handling for potential issues. Each operation is accompanied by code snippets to demonstrate the implementation process.

Uploaded by

Shabir Ahamed
Copyright
© © All Rights Reserved
We take content rights seriously. If you suspect this is your content, claim it here.
Available Formats
Download as DOCX, PDF, TXT or read online on Scribd
You are on page 1/ 8

CRUD Operations with Azure Cosmos DB in

Python
1. Initialize Cosmos DB Client
To interact with Cosmos DB, you need to initialize the
Cosmos client using your Cosmos DB endpoint and key.
Python
from azure.cosmos import CosmosClient

# Replace these with your actual Cosmos DB endpoint


and key
COSMOS_ENDPOINT = "https://<your-cosmosdb-
endpoint>:443/"
COSMOS_KEY = "<your-cosmosdb-key>"

# Initialize the Cosmos client


client = CosmosClient(COSMOS_ENDPOINT,
COSMOS_KEY)

# Access the database and container


database_name = "cosmicworks"
container_name = "products"

database = client.get_database_client(database_name)
container =
database.get_container_client(container_name)
2. Create (Insert) Operation

The create_item() method is used to insert a new


document into the container.
Insert a Single Item:
# Get database client
database = client.get_database_client(database_name)

# Get container client


container =
database.get_container_client(container_name)

# Define the item to insert


item_to_insert = {
"id": "1004", # The 'id' is mandatory and must be
unique
"categoryid": "my god", # Correctly formatted key-
value pair
"email": "shabir5490@gmail.com",
"age": 30
}

# Now, you can insert the item into Cosmos DB


try:
# Pass the item as the 'body' argument
container.create_item(body=item_to_insert)
print("Item inserted successfully!")
except CosmosHttpResponseError as e:
print(f"An error occurred while inserting the item:
{str(e)}")
# Insert a new item into the container
container.create_item(item_to_insert)
Insert Multiple Items:
You can insert multiple items by iterating over a list of
documents.
items_to_insert = [
{"id": "1", "name": "John Doe", "email":
"john.doe@example.com", "age": 30},
{"id": "2", "name": "Jane Smith", "email":
"jane.smith@example.com", "age": 25},
{"id": "3", "name": "Alice Johnson", "email":
"alice.johnson@example.com", "age": 28}
]
for item in items_to_insert:
container.create_item(item)

3. Read Operation
The read_item() method is used to read a document by
its id.
Read a Single Item by id:
# Get database client
database = client.get_database_client(database_name)

# Get container client


container =
database.get_container_client(container_name)

item_id = "986i"
partition_key = "my god"
item = container.read_item(item_id,
partition_key=partition_key)
print(item)

Query Items:
You can use SQL-like queries to retrieve documents
from the container.
try:
# Get database client
database = client.get_database_client(database_name)

# Get container client


container = database.get_container_client(container_name)

# Define the query


query = "SELECT * FROM c" # Example query,
replace with your actual query

# Fetch all items using the query, iterating over


results for efficiency
for item in container.query_items(query=query,
enable_cross_partition_query=True):
print(item)
except CosmosHttpResponseError as e:
# Log the error with more details
print(f"An error occurred while querying the items:
{e.message}")
print(f"Status code: {e.status_code}")
4. Update Operation
# Get database client
database = client.get_database_client(database_name)
# Get container client
container =
database.get_container_client(container_name)

try:
item_id = "986i"
partition_key = "my god"
# Step 1: Read the existing item
item = container.read_item(item=item_id,
partition_key=partition_key)
print(f"Original Item: {item}")

# Step 2: Update the desired fields


item["name"] = "cvduloah" # Change the name
field to "Abdullah"
item["email"]="cabdulkl4568s@gmail.coma"
# Step 3: Replace the item in the container
updated_item = container.replace_item(item=item,
body=item)
print(f"Updated Item: {updated_item}")

except CosmosHttpResponseError as e:
print(f"An error occurred while updating the item:
{str(e)}")

5. Delete Operation
The delete_item() method deletes a document by its id.
Delete a Single Item:
# Define the item id and partition key
item_id = "984i" # Example item id to delete
partition_key_value = "A0005" # Example partition key
value for the item

# Delete the item


try:
container.delete_item(item=item_id,
partition_key=partition_key_value)
print(f"Item with id {item_id} deleted successfully.")
except CosmosHttpResponseError as e:
print(f"An error occurred: {str(e)}")

6. Error Handling
You can handle Cosmos DB errors using
exceptions.CosmosHttpResponseError.
Python
Copy code
from Azure. cosmos import exceptions

try:
# Code that interacts with Cosmos DB
container.create_item(item_to_insert)
except for exceptions.CosmosHttpResponseError as e:
print(f"An error occurred: {str(e)}")

from azure.cosmos import CosmosClient


from azure.cosmos.exceptions import
CosmosHttpResponseError

# Replace these with your actual Cosmos DB endpoint


and key
COSMOS_ENDPOINT =
"https://cosmodb255.documents.azure.com:443/"
COSMOS_KEY =
"FBLWIQC9kZJB2AxHRUW3IWP4UKo6zb4o6uqnJBIZXn8z
RXRDZuxhtcyARekO0LuRIKY0CAd64DAUACDbJIZepQ==
"

# Initialize the Cosmos client


client = CosmosClient(COSMOS_ENDPOINT,
COSMOS_KEY)

# Access the database and container


database_name = "cosmicworks"
container_name = "products"

database = client.get_database_client(database_name)
container =
database.get_container_client(container_name)

# Item details to update


item_id = "986i" # The unique ID of the item
partition_key_value = "my god" # Assuming
`categoryid` is the partition key

try:
# Step 1: Read the existing item
item = container.read_item(item=item_id,
partition_key=partition_key_value)
print(f"Original Item: {item}")

# Step 2: Update the desired fields


item["name"] = "Abdullah" # Change the name field
to "Abdullah"
item["email"]="abdulla4568s@gmail.coma"
# Step 3: Replace the item in the container
updated_item = container.replace_item(item=item,
body=item)
print(f"Updated Item: {updated_item}")

except CosmosHttpResponseError as e:
print(f"An error occurred while updating the item:
{str(e)}")

You might also like