[本文是预发行文档，可能会有所更改。]
在本文中，我们演示了一些示例代码，这些代码使用 SDK 来处理 Dataverse 数据和元数据。 在继续阅读本文之前，请务必先阅读 入门（预览 版）。
基本操作
下面是一些操作帐户表的示例代码。
from azure.identity import InteractiveBrowserCredential
from PowerPlatform.Dataverse.client import DataverseClient
base_url = "https://<myorg>.crm.dynamics.com"
client = DataverseClient(base_url=base_url, credential=InteractiveBrowserCredential())
# Create an account and set some properties (returns list[str] of new GUIDs)
account_id = client.create("account", {"name": "Acme, Inc.", "telephone1": "555-0100"})[0]
# Read an account
account = client.get("account", account_id)
# Update an account (returns None)
client.update("account", account_id, {"telephone1": "555-0199"})
# Delete an account
client.delete("account", account_id)
批量操作
下面我们演示了几个执行批量更新的示例。
# Bulk update (broadcast) – apply same patch to several IDs
ids = client.create("account", [
{"name": "Contoso"},
{"name": "Fabrikam"},
])
client.update("account", ids, {"telephone1": "555-0200"}) # broadcast patch
# Bulk update (1:1) – list of patches matches list of IDs
client.update("account", ids, [
{"telephone1": "555-1200"},
{"telephone1": "555-1300"},
])
print({"multi_update": "ok"})
下面显示了一个创建多个帐户的示例。 传递有效负载列表到 create(logical_name, payloads) 以调用集合绑定的 Microsoft.Dynamics.CRM.CreateMultiple 操作。 该方法返回 list[str] 已创建的记录 ID。
# Bulk create accounts (returns list of GUIDs)
payloads = [
{"name": "Contoso"},
{"name": "Fabrikam"},
{"name": "Northwind"},
]
ids = client.create("account", payloads)
assert isinstance(ids, list) and all(isinstance(x, str) for x in ids)
print({"created_ids": ids})
有关批量操作的信息：
- 返回
None（与单个更新相同）以保持语义一致。 - 广播模式与按记录模式取决于
changes参数是字典还是列表。 - 构造 UpdateMultiple 操作 目标时，主键属性会被自动注入。
- 若有效负载遗漏 @odata.type，系统将自动添加（缓存逻辑名称查找）。
- 响应仅包括 ID - SDK 返回这些 GUID 字符串。
- 单记录创建返回包含单个元素的 GUID 列表。
- 每个实体集中的
@odata.type元数据查找被执行一次，结果缓存于内存中。
文件上载
下面是将名为“test.pdf”的文件上传到帐户记录的“sample_filecolumn” 的文件列 的几个示例。 第一个示例用于文件大小小于 128 MB，第二个示例用于文件大小超过 128 MB。
client.upload_file('account', record_id, 'sample_filecolumn', 'test.pdf')
client.upload_file('account', record_id, 'sample_filecolumn', 'test.pdf', mode='chunk', if_none_match=True)
有关文件上传的其他信息：
-
upload_file根据文件大小选取三种方法之一。 如果文件大小小于 128 MB，SDK 使用upload_file_small，否则，SDK 使用upload_file_chunk -
upload_file_small进行单个 Web API 调用，仅支持文件大小 < 128 MB。 -
upload_file_chunk将 PATCH 与 Content-Range 配合使用来上传文件（与 Dataverse 消息相比，与 HTTP 标准更一致）。 它由两个阶段组成 - 1。 通过 PATCH 请求获取实际上传使用的头部信息，以及 2。 实际按分块上传。 该函数使用在第一阶段返回的 ODatax-ms-chunk-size来确定区块大小（通常为 4 MB），然后使用Content-Range和Content-Length作为上传的元数据。 Web API 调用的总数是区块数 + 1。
分页批量检索
使用 get 函数逐页流式传输结果。 可以使用 $top 参数限制总结果，并使用 page_size 参数提示每页大小。 SDK 在内部设置 OData 首选标头 odata.maxpagesize。
pages = client.get(
"account",
select=["accountid", "name", "createdon"],
orderby=["name asc"],
top=10, # stop after 10 total rows (optional)
page_size=3, # ask for ~3 per page (optional)
)
total = 0
for page in pages: # each page is a list[dict]
print({"page_size": len(page), "sample": page[:2]})
total += len(page)
print({"total_rows": total})
支持的参数列表如下，除logical_name外其他所有参数都是可选的。
- logical_name：str — 逻辑名称（单一名称），例如“account”。
- select: list[str] | None — 列 -> $select（逗号分隔）。
- filter: str | None — OData $filter表达式（例如，contains(name,'Acme')和statecode eq 0）。
- orderby： list[str] |无 — 排序表达式 -> $orderby （逗号联接）。
- top：int | 无 — 通过 $top 设置全局上限（应用于第一个请求，服务在跨页面时强制执行）。
- expand: list[str] | None — 导航扩展 -> $expand; 传递原始子句（例如 primarycontactid($select=fullname,emailaddress1)）。
- page_size：int |无 — 使用 Prefer 的每页提示：odata.maxpagesize=<N> （不保证;最后一页可能更小）。
下面是返回值和语义的列表。
、 、 、 、 直接映射到与第一个请求的对应 OData 查询选项。 -
$top总行数上限;服务可能会跨多个页面对这些行进行分区。 - page_size （
Prefer: odata.maxpagesize） 是提示;服务器决定实际的页面边界。 - 返回生成无空页的生成器（list[dict]）。 跳过空页。
- 每个生成列表对应于 Web API 中的值页。
- 迭代在没有@odata.nextLink剩余或
$top在服务器端得到满足时停止。 - 生成器不会具体化所有结果;页面被延迟提取。
让我们看看一个示例，其中包含所有受支持的参数和预期响应。
pages = client.get(
"account",
select=["accountid", "name", "createdon", "primarycontactid"],
filter="contains(name,'Acme') and statecode eq 0",
orderby=["name asc", "createdon desc"],
top=5,
expand=["primarycontactid($select=fullname,emailaddress1)"],
page_size=2,
)
for page in pages: # page is list[dict]
# Expected page shape (illustrative)
# [
# {
# "accountid": "00000000-0000-0000-0000-000000000001"
# "name": "Acme West"
# "createdon": "2025-08-01T12:34:56Z"
# "primarycontactid": {
# "contactid": "00000000-0000-0000-0000-0000000000aa"
# "fullname": "Jane Doe"
# "emailaddress1": "<jane@acme.com>"
# }
# "@odata.etag": "W/\"123456\""
# }
#
# ]
print({"page_size": len(page)})
表元数据
让我们看看使用自定义表的示例代码。
# Support enumerations with labels in different languages
class Status(IntEnum):
Active = 1
Inactive = 2
Archived = 5
__labels__ = {
1033: {
"Active": "Active",
"Inactive": "Inactive",
"Archived": "Archived",
},
1036: {
"Active": "Actif",
"Inactive": "Inactif",
"Archived": "Archivé",
}
}
# Create a simple custom table and a few columns
info = client.create_table(
"SampleItem", # friendly name; defaults to SchemaName new_SampleItem
{
"code": "string",
"count": "int",
"amount": "decimal",
"when": "datetime",
"active": "bool",
"status": Status,
},
)
logical = info["entity_logical_name"] # for example, "new_sampleitem"
# Create a record in the new table
# Set your publisher prefix (used when creating the table). If you used the default, it's "new".
prefix = "new"
name_attr = f"{prefix}_name"
id_attr = f"{logical}id"
rec_id = client.create(logical, {name_attr: "Sample A"})[0]
# Clean up
client.delete(logical, rec_id) # delete record
client.delete_table("SampleItem") # delete table (friendly name or explicit schema new_SampleItem)
有关使用自定义表元数据的其他信息：
-
create始终返回 GUID 列表（单条输入时长度=1）。 -
update和delete对于单个接口和多个接口返回None。 - 传递一个负载列表到
create会触发批量创建，并返回一个包含ID的字符串列表。 -
get支持通过记录 ID 检索单条记录或分页浏览结果集（建议使用 select 限制列）。 - 对于采用记录 ID 的 CRUD 方法，请传递一个由连字符分隔的 36 个字符组成的 GUID 字符串。 GUID 周围的括号是允许的，但不是必须的。
- 使用
?sql=参数直接对实体集终端点执行SQL查询。 只支持子集（单个 SELECT，可选的 WHERE/TOP/ORDER BY 和别名）。 服务拒绝不受支持的构造。
将 pandas 与 SDK 配合使用
PandasODataClient 是一个封装低级客户端的轻量级封装器。 所有方法均接受逻辑（单数）名称（例如帐户、new_sampleitem），而不接受实体集（复数）名称，并且都受到支持。 有关工作流，请参阅名为“quickstart_pandas.py”的 SDK 源存储库DataFrame。