Microsoft Fabric 中的 Cosmos DB 完全依賴 Microsoft Entra ID 認證及內建的資料平面角色來管理認證與授權。 在本指南中，你會使用 Microsoft Entra ID 和已登入的帳號，連接到 Microsoft Fabric 中的 Cosmos DB 元件。
先決條件
現有的 Fabric 架構容量
- 如果您沒有 Fabric 容量， 請啟動 Fabric 試用版。
Fabric 中現有的 Cosmos DB 資料庫
- 如果您還沒有，請 在 Fabric 中建立新的 Cosmos DB 資料庫。
在 Fabric 中擁有 資料庫讀取 權限的身份
- 欲了解更多 Fabric 權限資訊，請參閱 存取控制。
取回 Cosmos DB 端點
首先，取得 Fabric 中 Cosmos DB 資料庫的端點。 此端點必須使用 Azure SDK 連接。
開啟 Fabric 入口網站 （https://app.fabric.microsoft.com）。
流覽至您現有的 Cosmos DB 資料庫。
在資料庫選單列中選擇 設定 選項。
在設定對話框中，進入 連線 區塊。 接著，複製 Endpoint for Cosmos DB NoSQL 資料庫 欄位的值。 這個值會在後續步驟中使用。
從筆記本進行驗證
Fabric 筆記本無法建立 Azure 服務所需的有效 Azure 憑證物件。 使用者使用 Fabric NotebookUtils 憑證工具 進行驗證，該工具需要 Cosmos DB 的認證範圍，並產生字串格式的憑證。 這需要一些自訂程式碼來建立有效的 Azure 憑證物件。
這裡有一個筆記本範例，示範如何做到這點。
儲存格 [1]：
#Install packages
%pip install azure-cosmos
細胞 [2]：
#Imports and config values
import logging
from azure.cosmos.aio import CosmosClient
from azure.cosmos.exceptions import CosmosHttpResponseError
COSMOS_ENDPOINT = 'https://my-cosmos-endpoint.cosmos.fabric.microsoft.com:443/'
COSMOS_DATABASE_NAME = '{your-cosmos-artifact-name}'
COSMOS_CONTAINER_NAME = '{your-container-name}'
Cell 3 的內容建立了 FabricTokenCredential() 物件，並包含必要的程式碼，以生成 Cosmos DB SDK 所需的有效憑證物件，這些憑證物件是從 Fabric NotebookUtils 憑證工具 產生的憑證字串製作的，以便驗證用戶身份。
儲存格 [3]：
# Custom TokenCredential implementation for Fabric authentication in a notebook
%pip install azure-core
from azure.core.credentials import TokenCredential, AccessToken
import base64
import json
import notebookutils
from datetime import datetime, timezone
class FabricTokenCredential(TokenCredential):
def get_token(self, *scopes: str, claims: Optional[str] = None, tenant_id: Optional[str] = None,
enable_cae: bool = False, **kwargs: Any) -> AccessToken:
access_token = notebookutils.credentials.getToken("https://cosmos.azure.com/")
parts = access_token.split(".")
if len(parts) < 2:
raise ValueError("Invalid JWT format")
payload_b64 = parts[1]
# Fix padding
padding = (-len(payload_b64)) % 4
if padding:
payload_b64 += "=" * padding
payload_json = base64.urlsafe_b64decode(payload_b64.encode("utf-8")).decode("utf-8")
payload = json.loads(payload_json)
exp = payload.get("exp")
if exp is None:
raise ValueError("exp claim missing in token")
return AccessToken(token=access_token, expires_on=exp)
細胞 [4]：
# Create a Cosmos client with the FabricTokenCredential() and run a query
async with CosmosClient(endpoint, credential=FabricTokenCredential()) as client:
container = client.get_database_client(COSMOS_DATABASE_NAME).get_container_client(COSMOS_CONTAINER_NAME)
query_text = "SELECT * FROM c"
results = container.query_items(query=query_text)
items = []
async for item in results:
items.append(item)
for item in items:
print(item)