Azure Cosmos DB TypeScript SDK
@azure/cosmos (TypeScript/JavaScript)
用于 Azure Cosmos DB NoSQL API 操作的数据平面 SDK —— 支持文档的 CRUD、查询和批量操作。
> ⚠️ 数据平面 vs 管理平面
> - 本 SDK (@azure/cosmos):针对文档、查询、存储过程的 CRUD 操作
> - 管理 SDK (@azure/arm-cosmosdb):通过 ARM 创建账户、数据库和容器
安装
npm install @azure/cosmos @azure/identity当前版本: 4.9.0
Node.js: >= 20.0.0
环境变量
COSMOS_ENDPOINT=https://<account>.documents.azure.com:443/
COSMOS_DATABASE=<database-name>
COSMOS_CONTAINER=<container-name>
仅用于密钥认证(推荐使用 AAD)
COSMOS_KEY=<account-key>身份验证
使用 DefaultAzureCredential 的 AAD (推荐)
import { CosmosClient } from "@azure/cosmos";
import { DefaultAzureCredential } from "@azure/identity";
const client = new CosmosClient({
endpoint: process.env.COSMOS_ENDPOINT!,
aadCredentials: new DefaultAzureCredential(),
});
基于密钥的身份验证
import { CosmosClient } from "@azure/cosmos";
// 选项 1:端点 + 密钥
const client = new CosmosClient({
endpoint: process.env.COSMOS_ENDPOINT!,
key: process.env.COSMOS_KEY!,
});
// 选项 2:连接字符串
const client = new CosmosClient(process.env.COSMOS_CONNECTION_STRING!);
资源层级
CosmosClient
└── Database (数据库)
└── Container (容器)
├── Items (项目/文档)
├── Scripts (脚本:存储过程、触发器、UDF)
└── Conflicts (冲突)核心操作
数据库与容器设置
const { database } = await client.databases.createIfNotExists({
id: "my-database",
});
const { container } = await database.containers.createIfNotExists({
id: "my-container",
partitionKey: { paths: ["/partitionKey"] },
});
创建文档
interface Product {
id: string;
partitionKey: string;
name: string;
price: number;
}
const item: Product = {
id: "product-1",
partitionKey: "electronics",
name: "Laptop",
price: 999.99,
};
const { resource } = await container.items.create<Product>(item);
读取文档
const { resource } = await container
.item("product-1", "electronics") // id, partitionKey
.read<Product>();
if (resource) {
console.log(resource.name);
}
更新文档 (替换)
const { resource: existing } = await container
.item("product-1", "electronics")
.read<Product>();
if (existing) {
existing.price = 899.99;
const { resource: updated } = await container
.item("product-1", "electronics")
.replace<Product>(existing);
}
Upsert 文档 (更新或插入)
const item: Product = {
id: "product-1",
partitionKey: "electronics",
name: "Laptop Pro",
price: 1299.99,
};
const { resource } = await container.items.upsert<Product>(item);
删除文档
await container.item("product-1", "electronics").delete();Patch 文档 (部分更新)
import { PatchOperation } from "@azure/cosmos";
const operations: PatchOperation[] = [
{ op: "replace", path: "/price", value: 799.99 },
{ op: "add",
path: "/discount", value: true },
{ op: "remove", path: "/oldField" },
];
const { resource } = await container
.item("product-1", "electronics")
.patch<Product>(operations);
## 查询 (Queries)
简单查询
### 参数化查询(推荐)const querySpec: SqlQuerySpec = {
query: "SELECT * FROM c WHERE c.partitionKey = @category AND c.price < @maxPrice",
parameters: [
{ name: "@category", value: "electronics" },
{ name: "@maxPrice", value: 1000 },
],
};
const { resources } = await container.items
.query<Product>(querySpec)
.fetchAll();
### 分页查询const queryIterator = container.items.query<Product>(querySpec, {
maxItemCount: 10, // 每页条数
});
while (queryIterator.hasMoreResults()) {
const { resources, continuationToken } = await queryIterator.fetchNext();
console.log(Page with ${resources?.length} items);
// 如有需要,使用 continuationToken 获取下一页
}
### 跨分区查询const { resources } = await container.items
.query<Product>(
"SELECT * FROM c WHERE c.price > 500",
{ enableCrossPartitionQuery: true }
)
.fetchAll();
## 批量操作 (Bulk Operations)
执行批量操作
const operations: OperationInput[] = [
{
operationType: BulkOperationType.Create,
resourceBody: { id: "1", partitionKey: "cat-a", name: "Item 1" },
},
{
operationType: BulkOperationType.Upsert,
resourceBody: { id: "2", partitionKey: "cat-a", name: "Item 2" },
},
{
operationType: BulkOperationType.Read,
id: "3",
partitionKey: "cat-b",
},
{
operationType: BulkOperationType.Replace,
id: "4",
partitionKey: "cat-b",
resourceBody: { id: "4", partitionKey: "cat-b", name: "Updated" },
},
{
operationType: BulkOperationType.Delete,
id: "5",
partitionKey: "cat-c",
},
{
operationType: BulkOperationType.Patch,
id: "6",
partitionKey: "cat-c",
resourceBody: {
operations: [{ op: "replace", path: "/name", value: "Patched" }],
},
},
];
const response = await container.items.executeBulkOperations(operations);
response.forEach((result, index) => {
if (result.statusCode >= 200 && result.statusCode < 300) {
console.log(Operation ${index} succeeded);
} else {
console.error(Operation ${index} failed: ${result.statusCode});
}
});
## 分区键 (Partition Keys)
简单分区键
### 分层分区键 (MultiHash)const { container } = await database.containers.createIfNotExists({
id: "orders",
partitionKey: {
paths: ["/tenantId", "/userId", "/sessionId"],
version: PartitionKeyDefinitionVersion.V2,
kind: PartitionKeyKind.MultiHash,
},
});
// 操作时需要提供分区键值数组
const { resource } = await container.items.create({
id: "order-1",
tenantId: "tenant-a",
userId: "user-123",
sessionId: "session-xyz",
total: 99.99,
});
// 使用分层分区键进行读取
const {
resource: order } = await container
.item("order-1", ["tenant-a", "user-123", "session-xyz"])
.read();
## 错误处理import { ErrorResponse } from "@azure/cosmos";
try {
const { resource } = await container.item("missing", "pk").read();
} catch (error) {
if (error instanceof ErrorResponse) {
switch (error.code) {
case 404:
console.log("未找到文档");
break;
case 409:
console.log("冲突 - 文档已存在");
break;
case 412:
console.log("前提条件失败 (ETag 不匹配)");
break;
case 429:
console.log("触发速率限制 - 请在以下时间后重试:", error.retryAfterInMs);
break;
default:
console.error(Cosmos 错误 ${error.code}: ${error.message});
}
}
throw error;
}
## 乐观并发控制 (ETags)// 读取 ETag
const { resource, etag } = await container
.item("product-1", "electronics")
.read<Product>();
if (resource && etag) {
resource.price = 899.99;
try {
// 仅在 ETag 匹配时进行替换
await container.item("product-1", "electronics").replace(resource, {
accessCondition: { type: "IfMatch", condition: etag },
});
} catch (error) {
if (error instanceof ErrorResponse && error.code === 412) {
console.log("文档已被另一个进程修改");
}
}
}
## TypeScript 类型参考import {
// 客户端与资源
CosmosClient,
Database,
Container,
Item,
Items,
// 操作
OperationInput,
BulkOperationType,
PatchOperation,
// 查询
SqlQuerySpec,
SqlParameter,
FeedOptions,
// 分区键
PartitionKeyDefinition,
PartitionKeyDefinitionVersion,
PartitionKeyKind,
// 响应
ItemResponse,
FeedResponse,
ResourceResponse,
// 错误
ErrorResponse,
} from "@azure/cosmos";
## 最佳实践
1. 使用 AAD 身份验证 — 优先使用 DefaultAzureCredential 而非密钥
2. 始终使用参数化查询 — 防止注入,提高执行计划缓存命中率
3. 指定分区键 — 尽可能避免跨分区查询
4. 使用批量操作 — 对于多次写入,使用 executeBulkOperations
5. 处理 429 错误 — 实现带有指数退避机制的重试逻辑
6. 使用 ETag 处理并发 — 防止并发场景下的更新丢失
7. 在关闭时关闭客户端 — 在清理阶段调用 client.dispose()
常见模式
服务层模式 (Service Layer Pattern)
constructor(client: CosmosClient) {
this.container = client
.database(process.env.COSMOS_DATABASE!)
.container(process.env.COSMOS_CONTAINER!);
}
async getById(id: string, category: string): Promise<Product | null> {
try {
const { resource } = await this.container
.item(id, category)
.read<Product>();
return resource ?? null;
} catch (error) {
if (error instanceof ErrorResponse && error.code === 404) {
return null;
}
throw error;
}
}
async create(product: Omit<Product, "id">): Promise<Product> {
const item = { ...product, id: crypto.randomUUID() };
const { resource } = await this.container.items.create<Product>(item);
return resource!;
}
async findByCategory(category: string): Promise<Product[]> {
const querySpec: SqlQuerySpec = {
query: "SELECT * FROM c WHERE c.partitionKey = @cat
egory",
parameters: [{ name: "@category", value: category }],
};
const { resources } = await this.container.items
.query<Product>(querySpec)
.fetchAll();
return resources;
}
}
``
相关 SDK
| SDK | 用途 | 安装命令 |
|-----|---------|---------|
| @azure/cosmos | 数据平面 (本 SDK) | npm install @azure/cosmos |@azure/arm-cosmosdb
| | 管理平面 (ARM) | npm install @azure/arm-cosmosdb |@azure/identity
| | 身份验证 | npm install @azure/identity` |
适用场景
本技能适用于执行概览中所描述的工作流或操作。局限性
- 仅在任务明确符合上述范围时使用此技能。
- 不要将输出结果视为针对特定环境的验证、测试或专家评审的替代方案。
- 如果缺少必要的输入、权限、安全边界或成功标准,请停止操作并请求澄清。