Azure Cosmosdb

参考: Azure Cosmos DB REST API — Authorization

解決策: azure-cosmos Python SDK を使用する。SDK 内部で DefaultAzureCredential をサポートしており、RBAC ロール（Cosmos DB Built-in Data Contributor 等）が付与されていれば認証が通る。

# NG: curl + Bearer token → 必ず HTTP 401
curl -H "Authorization: Bearer $(az account get-access-token --query accessToken -o tsv)" ...

# OK: azure-cosmos SDK + DefaultAzureCredential
python3 - <<'PYEOF'
from azure.cosmos import CosmosClient
from azure.identity import DefaultAzureCredential
client = CosmosClient(endpoint, credential=DefaultAzureCredential())
...
PYEOF

§2 SDK セットアップ（pip install バージョン指定）

インストールコマンド

pip install "azure-cosmos>=4.7,<5" "azure-identity>=1.16,<2" --quiet

• バージョン範囲指定を必須とする（無指定では破壊的変更の取り込みリスクがある）
• --quiet でログを抑制（CI ログ汚染防止）

スクリプト先頭での確認パターン

if ! python3 -c "import azure.cosmos, azure.identity" 2>/dev/null; then
  log_info "azure-cosmos / azure-identity をインストールします..."
  pip install "azure-cosmos>=4.7,<5" "azure-identity>=1.16,<2" --quiet || {
    log_error "pip install が失敗しました"
    exit 1
  }
fi

§3 Shell から Python SDK を呼び出すパターン（quoted heredoc）

ルール

1. quoted heredoc <<'PYEOF' を使用する（unquoted <<PYEOF はシェル変数が展開されインジェクションリスクがある）
2. シェル変数は export PY_* として環境変数でPython に渡す（heredoc 内への直接埋め込みは禁止）
3. python3 - で stdin から読む

テンプレート

# シェル変数を PY_ プレフィックスで export
export PY_COSMOS_ENDPOINT="https://${COSMOS_ACCOUNT}.documents.azure.com"
export PY_COSMOS_DB="${COSMOS_DB}"
export PY_COSMOS_CONTAINER="${COSMOS_CONTAINER}"
export PY_RESULT_FILE="${RESULT_FILE}"

python3 - <<'PYEOF'
import os
from azure.cosmos import CosmosClient
from azure.identity import DefaultAzureCredential

endpoint   = os.environ["PY_COSMOS_ENDPOINT"]
db_name    = os.environ["PY_COSMOS_DB"]
cont_name  = os.environ["PY_COSMOS_CONTAINER"]

cred      = DefaultAzureCredential()
client    = CosmosClient(endpoint, credential=cred)
db        = client.get_database_client(db_name)
container = db.get_container_client(cont_name)

# ... 処理 ...
PYEOF
PYRC=$?
[ "${PYRC}" -eq 0 ] || { log_error "Python SDK 呼び出しが失敗しました (exit ${PYRC})"; exit 1; }

§4 ドキュメント登録パターン（upsert + 件数検証 + 結果ファイル書き出し）

完全テンプレート

# パラメータ
COSMOS_ACCOUNT="${COSMOS_ACCOUNT:-<account-name>}"
COSMOS_DB="${COSMOS_DB:-<database-name>}"
COSMOS_CONTAINER="${COSMOS_CONTAINER:-<container-name>}"
DOCS_DIR="/tmp/<service>-samples"
RESULT_FILE="${TMPDIR:-/tmp}/<service>-reg-result.txt"
EXPECTED=3  # 登録期待件数（shell 側のみで管理。Python 結果ファイルには含めない）

# 環境変数として export
export PY_COSMOS_ENDPOINT="https://${COSMOS_ACCOUNT}.documents.azure.com"
export PY_COSMOS_DB="${COSMOS_DB}"
export PY_COSMOS_CONTAINER="${COSMOS_CONTAINER}"
export PY_DOCS_DIR="${DOCS_DIR}"
export PY_RESULT_FILE="${RESULT_FILE}"

python3 - <<'PYEOF'
import json, sys, os, glob, datetime
from azure.cosmos import CosmosClient, exceptions
from azure.identity import DefaultAzureCredential

endpoint   = os.environ["PY_COSMOS_ENDPOINT"]
db_name    = os.environ["PY_COSMOS_DB"]
cont_name  = os.environ["PY_COSMOS_CONTAINER"]
docs_dir   = os.environ["PY_DOCS_DIR"]
result_file = os.environ["PY_RESULT_FILE"]

cred      = DefaultAzureCredential()
client    = CosmosClient(endpoint, credential=cred)
container = client.get_database_client(db_name).get_container_client(cont_name)

inserted = 0
failed   = 0

for doc_file in sorted(glob.glob(os.path.join(docs_dir, "*.json"))):
    with open(doc_file, "r", encoding="utf-8") as f:
        doc = json.load(f)
    doc_id    = doc.get("id")
    partition = doc.get("<partition-key-field>")  # フィールド名は実装に合わせる
    if not doc_id:
        print(f"[ERROR] id が空 ({doc_file})", file=sys.stderr)
        failed += 1
        continue
    if not partition:
        print(f"[ERROR] partition key が空 ({doc_file})", file=sys.stderr)
        failed += 1
        continue
    try:
        container.upsert_item(doc)
        print(f"[INFO]  [OK] id={doc_id}")
        inserted += 1
    except exceptions.CosmosHttpResponseError as e:
        print(f"[ERROR] [FAIL] id={doc_id}: {e.status_code} {e.message}", file=sys.stderr)
        failed += 1

# 結果ファイルに書き出す（verify スクリプトが参照）
# EXPECTED は shell 側で管理するため、結果ファイルには書き出さない
ts = datetime.datetime.utcnow().strftime("%Y-%m-%dT%H:%M:%SZ")
with open(result_file, "w", encoding="utf-8") as rf:
    rf.write(f"SERVICE=CosmosDB\n")
    rf.write(f"CONTAINER={cont_name}\n")
    rf.write(f"INSERTED={inserted}\n")
    rf.write(f"FAILED={failed}\n")
    rf.write(f"TIMESTAMP={ts}\n")

print(f"\n登録済み: {inserted}、失敗: {failed}")

if failed > 0:
    # INSERTED < EXPECTED の最終判定は shell 側で行う
    print(f"[WARN]  {failed} 件失敗", file=sys.stderr)
sys.exit(0)
PYEOF
PYRC=$?
[ "${PYRC}" -eq 0 ] || { log_error "Python SDK 登録処理が失敗しました"; exit 1; }

# 結果ファイルから INSERTED を読み取って最終判定
INSERTED=$(grep '^INSERTED=' "${RESULT_FILE}" 2>/dev/null | cut -d= -f2 | tr -d '[:space:]' || echo 0)
if [ "${INSERTED}" -ge "${EXPECTED}" ]; then
  log_info "[OK] 登録件数 ${INSERTED}/${EXPECTED} 件"
else
  log_error "[FAIL] 登録件数 ${INSERTED}/${EXPECTED} 件（期待値未達）"
  exit 1  # 部分登録を成功扱いにしない
fi

重要なルール

§5 件数クエリパターン（COUNT(1) — verify スクリプト用フォールバック）

export PY_TC_ENDPOINT="https://${COSMOS_ACCOUNT}.documents.azure.com"
export PY_TC_DB="${COSMOS_DB}"
export PY_TC_CONTAINER="${COSMOS_CONTAINER}"

cosmos_count=$(python3 - <<'PYCOUNT'
import os, sys
from azure.cosmos import CosmosClient
from azure.identity import DefaultAzureCredential