Azure Cosmos Py

import osfrom azure.identity import DefaultAzureCredential, ManagedIdentityCredentialfrom azure.cosmos import CosmosClient
# Local dev: DefaultAzureCredential. Production: set AZURE_TOKEN_CREDENTIALS=prod or AZURE_TOKEN_CREDENTIALS=<specific_credential>credential = DefaultAzureCredential(require_envvar=True)# Or use a specific credential directly in production:# See https://learn.microsoft.com/python/api/overview/azure/identity-readme?view=azure-python#credential-classes# credential = ManagedIdentityCredential()
endpoint = "https://<account>.documents.azure.com:443/"
with CosmosClient(url=endpoint, credential=credential) as client:    # Use client here (see following sections for operations)    ...

# Get or create databasedatabase = client.create_database_if_not_exists(id="mydb")
# Get or create container with partition keycontainer = database.create_container_if_not_exists(    id="mycontainer",    partition_key=PartitionKey(path="/category"))
# Get existingdatabase = client.get_database_client("mydb")container = database.get_container_client("mycontainer")

item = {    "id": "item-001",           # Required: unique within partition    "category": "electronics",   # Partition key value    "name": "Laptop",    "price": 999.99,    "tags": ["computer", "portable"]}
created = container.create_item(body=item)print(f"Created: {created['id']}")

# Read requires id AND partition keyitem = container.read_item(    item="item-001",    partition_key="electronics")print(f"Name: {item['name']}")

item = container.read_item(item="item-001", partition_key="electronics")item["price"] = 899.99item["on_sale"] = True
updated = container.replace_item(item=item["id"], body=item)

# Create if not exists, replace if existsitem = {    "id": "item-002",    "category": "electronics",    "name": "Tablet",    "price": 499.99}
result = container.upsert_item(body=item)

container.delete_item(    item="item-001",    partition_key="electronics")

# Query within a partition (efficient)query = "SELECT * FROM c WHERE c.price < @max_price"items = container.query_items(    query=query,    parameters=[{"name": "@max_price", "value": 500}],    partition_key="electronics")
for item in items:    print(f"{item['name']}: ${item['price']}")

# Cross-partition (more expensive, use sparingly)query = "SELECT * FROM c WHERE c.price < @max_price"items = container.query_items(    query=query,    parameters=[{"name": "@max_price", "value": 500}],    enable_cross_partition_query=True)
for item in items:    print(item)

query = "SELECT c.id, c.name, c.price FROM c WHERE c.category = @category"items = container.query_items(    query=query,    parameters=[{"name": "@category", "value": "electronics"}],    partition_key="electronics")

# Read all in a partitionitems = container.read_all_items()  # Cross-partition# Or with partition keyitems = container.query_items(    query="SELECT * FROM c",    partition_key="electronics")

from azure.cosmos import PartitionKey
# Single partition keycontainer = database.create_container_if_not_exists(    id="orders",    partition_key=PartitionKey(path="/customer_id"))
# Hierarchical partition key (preview)container = database.create_container_if_not_exists(    id="events",    partition_key=PartitionKey(path=["/tenant_id", "/user_id"]))

# Create container with provisioned throughputcontainer = database.create_container_if_not_exists(    id="mycontainer",    partition_key=PartitionKey(path="/pk"),    offer_throughput=400  # RU/s)
# Read current throughputoffer = container.read_offer()print(f"Throughput: {offer.offer_throughput} RU/s")
# Update throughputcontainer.replace_throughput(throughput=1000)

from azure.cosmos.aio import CosmosClientfrom azure.identity.aio import DefaultAzureCredential
async def cosmos_operations():    async with DefaultAzureCredential() as credential:        async with CosmosClient(endpoint, credential=credential) as client:            database = client.get_database_client("mydb")            container = database.get_container_client("mycontainer")                        # Create            await container.create_item(body={"id": "1", "pk": "test"})                        # Read            item = await container.read_item(item="1", partition_key="test")                        # Query            async for item in container.query_items(                query="SELECT * FROM c",                partition_key="test"            ):                print(item)
import asyncioasyncio.run(cosmos_operations())

from azure.cosmos.exceptions import CosmosHttpResponseError
try:    item = container.read_item(item="nonexistent", partition_key="pk")except CosmosHttpResponseError as e:    if e.status_code == 404:        print("Item not found")    elif e.status_code == 429:        print(f"Rate limited. Retry after: {e.headers.get('x-ms-retry-after-ms')}ms")    else:        raise