AutoGineer
Generated banner for Azure Cosmos Py
Generated logo for Azure Cosmos Py

Azure Cosmos Py

MicrosoftOfficialCreated by Microsoft

  • Skill
  • Tool integration
  • Data and AI
  • MIT
  • Low risk
Download ZIP

From the author

Azure Cosmos DB SDK for Python (NoSQL API). Use for document CRUD, queries, containers, and globally distributed data. Triggers: "cosmos db", "CosmosClient", "container", "document", "NoSQL", "partition key".
— the description by Microsoft

AutoGineer summary

Written by AutoGineer from the skill’s own files, not by the author

The azure-cosmos-py skill provides the Python client library for Azure Cosmos DB NoSQL API. It enables developers to perform CRUD operations, manage databases and containers, execute queries, and handle throughput settings. The skill emphasizes using DefaultAzureCredential for authentication and context managers for resource lifecycle management, while providing guidance on partition key design and query optimization patterns.

What it does

  • Provides client classes for account, database, and container operations.
  • Supports document CRUD operations including create, read, replace, upsert, and delete.
  • Enables SQL-like querying of data with support for parameters and projections.
  • Facilitates management of container throughput and partition key configuration.
  • Offers both synchronous and asynchronous client implementations.
  • Includes patterns for pagination and handling hierarchical partition keys.

When to use it

  • When developing Python applications that need to interact with Azure Cosmos DB.
  • When implementing data access layers for NoSQL document storage.
  • When designing partition strategies for data distribution.
  • When optimizing query performance through parameterized queries and efficient partition key usage.

Advantages

  • Provides a unified interface for both sync and async operations.
  • Integrates with Azure identity services for credential management.
  • Includes built-in support for handling common database exceptions.
  • Offers clear patterns for managing resource lifecycles using context managers.

Usage examples

“How do I create a new item in an Azure Cosmos DB container using Python?”
The agent provides a code snippet using container.create_item() with a dictionary representing the item, ensuring the partition key is included.
“How can I query items in Cosmos DB while avoiding cross-partition overhead?”
The agent demonstrates how to use container.query_items() by specifying the partition_key parameter to target a single partition.
“What is the best way to handle authentication in my Cosmos DB Python code?”
The agent explains the use of DefaultAzureCredential and provides a code example showing how to pass it to the CosmosClient.
“How do I implement pagination for a large query result set?”
The agent provides a function using continuation tokens with query_items() to iterate through results page by page.

Install

  1. Download the skill as a ZIP:

    https://autogineer.com/agent-skills/microsoft/azure-cosmos-py/download

  2. Unzip it into the cross-client skills folder of your project:

    mkdir -p .agents/skills && unzip azure-cosmos-py.zip -d .agents/skills

Documentation by Microsoft

Azure Cosmos DB SDK for Python

Client library for Azure Cosmos DB NoSQL API — globally distributed, multi-model database.

Installation

pip install azure-cosmos azure-identity

Environment Variables

COSMOS_ENDPOINT=https://<account>.documents.azure.com:443/  # Required for all auth methods
COSMOS_DATABASE=mydb  # Required for all auth methods
COSMOS_CONTAINER=mycontainer  # Required for all auth methods
AZURE_TOKEN_CREDENTIALS=prod # Required only if DefaultAzureCredential is used in production

Authentication & Lifecycle

🔑 Two rules apply to every code sample below:

  1. Prefer DefaultAzureCredential. It works locally (Azure CLI / VS Code / Developer CLI) and in Azure (managed identity, workload identity) with no code change. Avoid connection strings, account/API keys — they bypass Entra audit and rotation.

    • Local dev: DefaultAzureCredential works as-is.

    • Production: set AZURE_TOKEN_CREDENTIALS=prod (or AZURE_TOKEN_CREDENTIALS=<specific_credential>) to constrain the credential chain to production-safe credentials.

  2. Wrap every client in a context manager so HTTP transports, sockets, and token caches are released deterministically:

    • Sync: with <Client>(...) as client:

    • Async: async with <Client>(...) as client: and async with DefaultAzureCredential() as credential: (from azure.identity.aio)

Snippets may abbreviate this setup, but production code should always follow both rules.

import os
from azure.identity import DefaultAzureCredential, ManagedIdentityCredential
from 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)
    ...

Client Hierarchy

ClientPurposeGet From
CosmosClientAccount-level operationsDirect instantiation
DatabaseProxyDatabase operationsclient.get_database_client()
ContainerProxyContainer/item operationsdatabase.get_container_client()

Core Workflow

Setup Database and Container
# Get or create database
database = client.create_database_if_not_exists(id="mydb")

# Get or create container with partition key
container = database.create_container_if_not_exists(
    id="mycontainer",
    partition_key=PartitionKey(path="/category")
)

# Get existing
database = client.get_database_client("mydb")
container = database.get_container_client("mycontainer")
Create Item
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 Item
# Read requires id AND partition key
item = container.read_item(
    item="item-001",
    partition_key="electronics"
)
print(f"Name: {item['name']}")
Update Item (Replace)
item = container.read_item(item="item-001", partition_key="electronics")
item["price"] = 899.99
item["on_sale"] = True

updated = container.replace_item(item=item["id"], body=item)
Upsert Item
# Create if not exists, replace if exists
item = {
    "id": "item-002",
    "category": "electronics",
    "name": "Tablet",
    "price": 499.99
}

result = container.upsert_item(body=item)
Delete Item
container.delete_item(
    item="item-001",
    partition_key="electronics"
)

Queries

Basic Query
# 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 Query
# 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 with Projection
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 Items
# Read all in a partition
items = container.read_all_items()  # Cross-partition
# Or with partition key
items = container.query_items(
    query="SELECT * FROM c",
    partition_key="electronics"
)

Partition Keys

Critical: Always include partition key for efficient operations.

from azure.cosmos import PartitionKey

# Single partition key
container = 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"])
)

Throughput

# Create container with provisioned throughput
container = database.create_container_if_not_exists(
    id="mycontainer",
    partition_key=PartitionKey(path="/pk"),
    offer_throughput=400  # RU/s
)

# Read current throughput
offer = container.read_offer()
print(f"Throughput: {offer.offer_throughput} RU/s")

# Update throughput
container.replace_throughput(throughput=1000)

Async Client

from azure.cosmos.aio import CosmosClient
from 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 asyncio
asyncio.run(cosmos_operations())

Error Handling

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

Best Practices

  1. Pick sync OR async and stay consistent. Do not mix azure.cosmos sync clients with azure.cosmos.aio async clients in the same call path. Choose one mode per module.

  2. Always use context managers for clients and async credentials. Wrap every client in with CosmosClient(...) as client: (sync) or async with CosmosClient(...) as client: (async). For async DefaultAzureCredential from azure.identity.aio, also use async with credential: so tokens and transports are cleaned up.

  3. Use DefaultAzureCredential for portable auth across local dev and Azure (avoid connection strings / API keys when possible).

  4. Always specify partition key for point reads and queries

  5. Use parameterized queries to prevent injection and improve caching

  6. Avoid cross-partition queries when possible

  7. Use upsert_item for idempotent writes

  8. Use async client for high-throughput scenarios

  9. Design partition key for even data distribution

  10. Use read_item instead of query for single document retrieval

Reference Files

FileContents
references/partitioning.mdPartition key strategies, hierarchical keys, hot partition detection and mitigation
references/query-patterns.mdQuery optimization, aggregations, pagination, transactions, change feed
scripts/setup_cosmos_container.pyCLI tool for creating containers with partitioning, throughput, and indexing

Files

Requirements and compatibility

Works with: Claude Code, Claude apps, OpenAI Codex, GitHub Copilot, Cursor, Gemini CLI, Kiro, OpenCode, Goose, Amp, Roo Code, JetBrains Junie, Factory Droid, Devin (ex-Windsurf), Mistral Vibe, OpenHands, Letta

“Works with” means the tool reads the open Agent Skills format; a skill written for one tool may use features others lack.

Needs: Needs an API key, Needs Python

Security

What our scanner noticed· Low risk

  • Instructions that use the network

A static scan is not a guarantee. Skills are third-party content: review a skill before you let an agent use it.

Attribution and licence

Created by
Microsoft
Publisher
Microsoft (Official)
Source
microsoft/skills · .github/plugins/azure-sdk-python/skills/azure-cosmos-py · commit 0696600
Licence
MIT License (redistribution allowed)
  • LICENSE (added by AutoGineer from LICENSE)
  • ATTRIBUTION.md (written by AutoGineer)
Synced

AutoGineer is not affiliated with or endorsed by Microsoft or Microsoft.

Download ZIP

The ZIP holds the skill’s files, its licence files and an ATTRIBUTION.md. Download azure-cosmos-py.zip