Microsoft Fabric 中的 Cosmos DB 完全依赖于 Microsoft Entra ID 身份验证和内置数据平面角色来管理身份验证和授权。 在本指南中，您将使用 Microsoft Entra ID 和已登录的帐户来连接到 Microsoft Fabric 中的 Cosmos DB 工件。
先决条件
现有Fabric容量
- 如果没有 Fabric 容量， 请启动 Fabric 试用版。
Fabric 中的现有 Cosmos DB 数据库
具有 Fabric 中数据库的 读取 权限的标识
- 有关 Fabric 权限的详细信息，请参阅 访问控制。
检索 Cosmos DB 终结点
首先，获取 Fabric 中 Cosmos DB 数据库的终结点。 此终结点是使用 Azure SDK 进行连接所必需的。
打开 Fabric 门户（https://app.fabric.microsoft.com）。
导航到现有的 Cosmos DB 数据库。
在数据库的菜单栏中选择 “设置” 选项。
在“设置”对话框中，导航到 “连接” 部分。 然后，复制 Cosmos DB NoSQL 数据库 字段的 Endpoint 值。 在后续步骤中使用此值。
从笔记本进行身份验证
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（） 对象，并包含从 Fabric NotebookUtils 凭据实用工具生成的令牌字符串中为 Cosmos DB SDK 生成有效凭据对象所需的代码，这是对用户进行身份验证所必需的。
单元格 [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)