azure-cosmos-db-py

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

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

復元されたヘルパーの安全性

統合テスト用フィクスチャ assets/conftest_template.py は、設定済みの Cosmos エンドポイントに接続します。cleanup_test_docs は、追跡リストに明示的に追加されたレコードを削除します。隔離された破棄可能なテストコンテナーのみを使用し、テスト専用のレコードを使ってください。本番データは決して使用しないでください。テンプレートをコピーしても、Cosmos には接続されません。

Cosmos DB サービスの実装

修正事項の確認（2026-10-23）：以下のスニペットは、不完全なアプリケーションのひな型です。イベントループの外で起動時に同期クライアントを初期化し、初期化を直列化したうえで、シャットダウン時にクライアントと資格情報を閉じてください。初期化に失敗した状態がキャッシュされた場合は、明示的で上限のある復旧経路が必要です。azure.cosmos.aio も利用できます。SDK は同期専用ではありません。パーティションのルーティングとは別に、テナント認可を適用してください。バンドルされたクライアントは /workspaceId を前提としています。呼び出し元から SQL の断片を受け入れてはいけません。ストレージが利用できない状態（503）と、見つからない状態（404）、データが空の状態を区別してください。以下の古い、デフォルト値を返して正常に処理を続ける例は、正確性が求められる読み取りでは避けるべきポリシーを示しています。更新の消失を防ぐために ETag 条件付き書き込みを使用し、一意性をアトミックに適用し、競合を考慮した補償処理によってパーティション間の移動を実装してください。読み取り後に upsert または削除を行う処理はアトミックではありません。これらのテンプレートでは、実際の Cosmos の動作は検証されていません。

クリーンコード、セキュリティのベストプラクティス、TDD の原則に従って、本番環境に対応した Azure Cosmos DB NoSQL サービスを構築してください。

インストール

pip install azure-cosmos azure-identity

環境変数

COSMOS_ENDPOINT=https://<account>.documents.azure.com:443/
COSMOS_DATABASE_NAME=<database-name>
COSMOS_CONTAINER_ID=<container-id>
# For emulator only (not production)
COSMOS_KEY=<emulator-key>

認証

DefaultAzureCredential（推奨）：

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

client = CosmosClient(
    url=os.environ["COSMOS_ENDPOINT"],
    credential=DefaultAzureCredential()
)

エミュレーター（ローカル開発）：

from azure.cosmos import CosmosClient

client = CosmosClient(
    url="https://localhost:8081",
    credential=os.environ["COSMOS_KEY"],
    connection_verify=False
)

アーキテクチャの概要

┌─────────────────────────────────────────────────────────────────┐
│                         FastAPI Router                          │
│  - Auth dependencies (get_current_user, get_current_user_required)
│  - HTTP error responses (HTTPException)                         │
└──────────────────────────────┬──────────────────────────────────┘
                               │
┌──────────────────────────────▼──────────────────────────────────┐
│                        Service Layer                            │
│  - Business logic and validation                                │
│  - Document ↔ Model conversion                                  │
│  - Graceful degradation when Cosmos unavailable                 │
└──────────────────────────────┬──────────────────────────────────┘
                               │
┌──────────────────────────────▼──────────────────────────────────┐
│                     Cosmos DB Client Module                     │
│  - Singleton container initialization                           │
│  - Dual auth: DefaultAzureCredential (Azure) / Key (emulator)   │
│  - Async wrapper via run_in_threadpool                          │
└─────────────────────────────────────────────────────────────────┘

クイックスタート

1. クライアントモジュールのセットアップ

二重認証に対応した Cosmos クライアントをシングルトンとして作成します：

# db/cosmos.py
from azure.cosmos import CosmosClient
from azure.identity import DefaultAzureCredential
from starlette.concurrency import run_in_threadpool

_cosmos_container = None

def _is_emulator_endpoint(endpoint: str) -> bool:
    from urllib.parse import urlsplit
    parsed = urlsplit(endpoint)
    return (parsed.scheme == "https" and parsed.username is None
            and parsed.password is None
            and parsed.hostname in {"localhost", "127.0.0.1", "::1"})

def get_container():
    global _cosmos_container
    if _cosmos_container is None:
        if _is_emulator_endpoint(settings.cosmos_endpoint):
            client = CosmosClient(
                url=settings.cosmos_endpoint,
                credential=settings.cosmos_key,
                connection_verify=False
            )
        else:
            client = CosmosClient(
                url=settings.cosmos_endpoint,
                credential=DefaultAzureCredential()
            )
        db = client.get_database_client(settings.cosmos_database_name)
        _cosmos_container = db.get_container_client(settings.cosmos_container_id)
    return _cosmos_container

完全な実装：references/client-setup.md を参照してください

2. Pydantic モデル階層

明確に分離するため、5 層のモデルパターンを使用します：

class ProjectBase(BaseModel):           # Shared fields
    name: str = Field(..., min_length=1, max_length=200)

class ProjectCreate(ProjectBase):       # Creation request
    workspace_id: str = Field(..., alias="workspaceId")

class ProjectUpdate(BaseModel):         # Partial updates (all optional)
    name: Optional[str] = Field(None, min_length=1)

class Project(ProjectBase):             # API response
    id: str
    created_at: datetime = Field(..., alias="createdAt")

class ProjectInDB(Project):             # Internal with docType
    doc_type: str = "project"

3. サービスレイヤーパターン

class ProjectService:
    def _use_cosmos(self) -> bool:
        return get_container() is not None
    
    async def get_by_id(self, project_id: str, workspace_id: str) -> Project | None:
        if not self._use_cosmos():
            return None
        doc = await get_document(project_id, partition_key=workspace_id)
        if doc is None:
            return None
        return self._doc_to_model(doc)

完全なパターン：references/service-layer.md を参照してください

基本原則

セキュリティ要件

1. RBAC 認証：Azure では DefaultAzureCredential を使用し、コードにキーを保存しないでください
2. エミュレーター専用キー：ローカル開発でのみ、既知のエミュレーターキーをハードコードしてください
3. パラメーター化クエリ：常に @parameter 構文を使用し、文字列の連結は決して行わないでください
4. パーティションキーの検証：パーティションキーへのアクセスがユーザーの認可と一致することを検証してください

クリーンコードの規約

1. 単一責任：クライアントモジュールは接続を処理し、サービスはビジネスロジックを処理します
2. グレースフルデグラデーション：Cosmos が利用できない場合、サービスは None／[] を返します
3. 一貫した命名：_doc_to_model()、_model_to_doc()、_use_cosmos()
4. 型ヒント：すべてのパブリックメソッドに完全な型を指定します
5. キャメルケースのエイリアス：JSON シリアル化には Field(alias="camelCase") を使用します

TDD の要件

以下のパターンを使用して、実装の前にテストを書いてください：

@pytest.fixture
def mock_cosmos_container(mocker):
    container = mocker.MagicMock()
    mocker.patch("app.db.cosmos.get_container", return_value=container)
    return container

@pytest.mark.asyncio
async def test_get_project_by_id_returns_project(mock_cosmos_container):
    # Arrange
    mock_cosmos_container.read_item.return_value = {"id": "123", "name": "Test"}
    
    # Act
    result = await project_service.get_by_id("123", "workspace-1")
    
    # Assert
    assert result.id == "123"
    assert result.name == "Test"

テストガイド全文：references/testing.md を参照してください

リファレンスファイル

ファイル	読むタイミング
references/client-setup.md	二重認証、SSL 設定、シングルトンパターンを使用した Cosmos クライアントのセットアップ
references/service-layer.md	CRUD、変換、グレースフルデグラデーションを備えた完全なサービスクラスの実装
references/testing.md	pytest テストの作成、Cosmos のモック、統合テストのセットアップ
references/partitioning.md	パーティションキーの選択、パーティション間クエリ、移動操作
references/error-handling.md	CosmosResourceNotFoundError の処理、ログ記録、HTTP エラーのマッピング

テンプレートファイル

ファイル	用途
assets/cosmos_client_template.py	すぐに使えるクライアントモジュール
assets/service_template.py	サービスクラスのひな型
assets/conftest_template.py	Cosmos のモック用 pytest フィクスチャ

品質属性（NFR）

信頼性

• Cosmos が利用できない場合でも、適切に機能を縮退します
• 一時的な障害に対する指数バックオフ付きの再試行ロジック
• シングルトンパターンによる接続プーリング

セキュリティ

• コード内にシークレットを含めません（DefaultAzureCredential による RBAC）
• パラメーター化クエリでインジェクションを防ぎます
• パーティションルーティングはテナントを認可しません。ワークスペース ID を受け入れる前に、メンバーシップ／ロールを確認してください

保守性

• 5 層モデルパターンによりスキーマの進化に対応します
• サービスレイヤーでビジネスロジックをストレージから分離します
• すべてのエンティティサービスで一貫したパターンを採用します

テスト容易性

• get_container() による依存性注入
• モジュールレベルのグローバル変数で簡単にモックできます
• 明確な分離により、Cosmos なしで単体テストを実施できます

パフォーマンス

• パーティションキーを使用したクエリで、全パーティションのスキャンを回避します
• 非同期ラッパーで FastAPI のイベントループのブロッキングを防ぎます
• ドキュメント変換のオーバーヘッドを最小限に抑えます

使用する場面

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

制限事項

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