azure-cosmos

import os from azure.cosmos import CosmosClient # Retrieve Cosmos DB account details from environment variables endpoint = os.environ.get('AZURE_COSMOS_ENDPOINT', 'YOUR_COSMOS_ENDPOINT') key = os.environ.get('AZURE_COSMOS_KEY', 'YOUR_COSMOS_KEY') # Ensure endpoint and key are not default placeholders if 'YOUR_COSMOS_ENDPOINT' in endpoint or 'YOUR_COSMOS_KEY' in key: print("Please set AZURE_COSMOS_ENDPOINT and AZURE_COSMOS_KEY environment variables.") exit(1) database_name = 'MyDatabase' container_name = 'MyContainer' client = CosmosClient(endpoint, key) try: # Create a database database = client.create_database_if_not_exists(id=database_name) print(f"Database '{database_name}' created or already exists.") # Create a container with a partition key container = database.create_container_if_not_exists( id=container_name, partition_key={'path': '/category', 'kind': 'Hash'} ) print(f"Container '{container_name}' created or already exists.") # Create an item new_item = {"id": "item1", "name": "Sample Item", "category": "Electronics", "price": 100} item = container.create_item(body=new_item) print(f"Item created with id: {item['id']}") # Read an item read_item = container.read_item(item=item['id'], partition_key=item['category']) print(f"Item read: {read_item['name']}") # Query items query = "SELECT * FROM c WHERE c.category = 'Electronics'" items = list(container.query_items(query=query, enable_cross_partition_query=True)) print(f"Found {len(items)} item(s) in category 'Electronics'.") finally: # Clean up (optional): delete database and its contents # client.delete_database(database.id) # print(f"Database '{database_name}' deleted.") pass