azure-cosmos-py

バンドルされた上流サポート

サポートファイルはスキル相対パスに含まれています。固定されたソース、ライセンス、ハッシュ、インポート時の変更については、UPSTREAM-NOTICE.md を参照してください。

サポート資料の利用可否

以下のパスは、このカタログのコピーでは引き続き利用できません。

• <support-root>/scripts/setup_cosmos_container.py

これらのファイルを必要とする手順は実行しないでください。使用する前に、オリジナルの資料を入手し、その依存関係、ライセンス、影響を確認してください。独立したインライン手順とバンドルされたヘルパーは、ユーザーが許可した範囲内で実行できます。

Azure Cosmos DB SDK for Python

Azure Cosmos DB NoSQL API 用のクライアントライブラリです。グローバルに分散されたマルチモデルデータベースです。

インストール

pip install azure-cosmos azure-identity

環境変数

COSMOS_ENDPOINT=https://<account>.documents.azure.com:443/
COSMOS_DATABASE=mydb
COSMOS_CONTAINER=mycontainer

認証

from azure.identity import DefaultAzureCredential
from azure.cosmos import CosmosClient

credential = DefaultAzureCredential()
endpoint = "https://<account>.documents.azure.com:443/"

client = CosmosClient(url=endpoint, credential=credential)

クライアント階層

クライアント	用途	取得元
CosmosClient	アカウントレベルの操作	直接インスタンス化
DatabaseProxy	データベースの操作	client.get_database_client()
ContainerProxy	コンテナー／項目の操作	database.get_container_client()

基本的なワークフロー

プロビジョニングやスループットの変更を行う前に、アカウント、データベース、コンテナー、パーティションキーの定義、および許可された予算を確認してください。create_*_if_not_exists は既存のコンテナー設定を調整しません。例では異なるキーのパスを使用しています。同じコンテナーに対してこれらを組み合わせないでください。テナントによるパーティション分割は、認可の分離にはなりません。読み取り・変更・書き込み操作には ETag 条件を使用し、再試行に備えて論理操作の重複排除を明示的に行ってください。変更フィードは、グローバルな順序や厳密に一度だけの処理を保証する台帳ではありません。冪等な処理が成功した後にのみチェックポイントを設定し、固定された SDK のチェックポイント API を確認してください。

データベースとコンテナーのセットアップ

from azure.cosmos import PartitionKey

# 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")

項目の作成

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 key
item = 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.99
item["on_sale"] = True

updated = container.replace_item(item=item["id"], body=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)

項目の削除

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 partition
items = container.read_all_items()  # Cross-partition
# Or with partition key
items = container.query_items(
    query="SELECT * FROM c",
    partition_key="electronics"
)

パーティションキー

重要：効率的な操作のため、必ずパーティションキーを指定してください。

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"])
)

スループット

# 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.get_throughput()
print(f"Throughput: {offer.offer_throughput} RU/s")

# Update throughput
container.replace_throughput(throughput=1000)

非同期クライアント

from azure.cosmos.aio import CosmosClient
from azure.identity.aio import DefaultAzureCredential

async def cosmos_operations():
    async with DefaultAzureCredential() as credential, 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())

エラー処理

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

ベストプラクティス

1. ポイント読み取りとクエリでは、必ずパーティションキーを指定してください
2. インジェクションを防ぎ、キャッシュを改善するため、パラメーター化クエリを使用してください
3. 可能な限り、パーティション間クエリを避けてください
4. 安定した ID は慎重に使用してください。アップサートは既存の内容を置き換えますが、並行性や論理操作の重複排除を保証するものではありません
5. 高スループットのシナリオでは、非同期クライアントを使用してください
6. データが均等に分散されるようにパーティションキーを設計してください
7. 単一のドキュメントを取得する場合は、クエリではなく read_item を使用してください

参照ファイル

ファイル	内容
references/partitioning.md	パーティションキーの戦略、階層キー、ホットパーティションの検出と緩和
references/query-patterns.md	クエリの最適化、集計、ページネーション、トランザクション、変更フィード
scripts/setup_cosmos_container.py	パーティション分割、スループット、インデックスを指定してコンテナーを作成する CLI ツール

使用する場合

このスキルは、概要で説明されているワークフローやアクションを実行する場合に適用できます。

制限事項

• このスキルは、タスクが上記の範囲に明確に一致する場合にのみ使用してください。
• 出力を、環境固有の検証、テスト、専門家によるレビューの代わりとして扱わないでください。
• 必要な入力、権限、安全上の境界、または成功基準が不足している場合は、作業を中止して確認を求めてください。