DataverseClient 類別

Microsoft Dataverse 營運的高階客戶端。

此用戶端提供簡單且穩定的介面，透過 Web API 與 Dataverse 環境互動。 它透過 Azure Identity 處理認證，並將 HTTP 操作委派給內部 <xref:PowerPlatform.Dataverse.data._odata._ODataClient>。

主要功能：

• OData CRUD 操作：建立、讀取、更新、刪除紀錄

• SQL 查詢：透過 Web API ?sql 參數執行唯讀 SQL

• 資料表元資料：建立、檢查及刪除自訂資料表;建立與刪除欄位

• 檔案上傳：將檔案上傳到支援分塊的檔案欄位

備註

用戶端在首次使用時會懶散地初始化內部 OData 用戶端，允許輕量級建置而無需立即呼叫網路。

建構函式

DataverseClient(base_url: str, credential: TokenCredential, config: DataverseConfig | None = None)

參數

名稱	Description
base_url 必要	str 例如 "https://org.crm.dynamics.com"你的 Dataverse 環境 URL。 尾部斬擊會自動移除。
credential 必要	TokenCredential Azure Identity credential for authentication.
config	DataverseConfig 或 None 語言、超時和重試的可選設定。 若未提供，預設值將從 載入 from_env。 預設值: None

範例

建立客戶端並執行基本操作：

   from azure.identity import InteractiveBrowserCredential
   from PowerPlatform.Dataverse.client import DataverseClient

   credential = InteractiveBrowserCredential()
   client = DataverseClient(
       "https://org.crm.dynamics.com",
       credential
   )

   # Create a record
   record_ids = client.create("account", {"name": "Contoso Ltd"})
   print(f"Created account: {record_ids[0]}")

   # Update a record
   client.update("account", record_ids[0], {"telephone1": "555-0100"})

   # Query records
   for batch in client.get("account", filter="name eq 'Contoso Ltd'"):
       for account in batch:
           print(account["name"])

   # Delete a record
   client.delete("account", record_ids[0])

方法

create	以資料表名稱建立一個或多個紀錄。
create_columns	利用結構式映射在現有資料表上建立一個或多個欄位。
create_table	建立一個簡單的自訂表格，並指定欄位。
delete	透過 GUID 刪除一個或多個紀錄。
delete_columns	從表格中刪除一個或多個欄位。
delete_table	以名稱刪除自訂資料表。 警告 此操作不可逆，會刪除表格中所有記錄 並定義了表格。 請小心使用。
flush_cache	清除快取的客戶端元資料或狀態。
get	可以依 ID 取得單一紀錄，或查詢多個紀錄。 提供時 record_id ，會回傳單一記錄字典。 當 record_id 為 None 時，回傳一個產生器，產生批量記錄。
get_table_info	如果有表格的基本元資料，請取得。
list_tables	列出 Dataverse 環境中所有自訂資料表。
query_sql	使用 Dataverse Web API ?sql 功能執行唯讀 SQL 查詢。 SQL 查詢必須遵循支援的子集：一個 SELECT 陳述式，並可選地使用 WHERE、TOP（整數字面量）、ORDER BY（僅限欄位名稱）以及 FIN 後的簡單表格別名。 備註 SQL 支援僅限於唯讀查詢。 複雜的連接、子查詢以及某些 SQL 函式可能不被支援。 請參閱 Dataverse 文件以了解目前的功能集。
update	更新一個或多個紀錄。 此方法支援三種使用模式： 單筆紀錄更新： update("account", "guid", {"name": "New Name"}) 廣播更新： update("account", [id1, id2], {"status": 1}) - 將相同的變更套用到所有 ID 配對更新： update("account", [id1, id2], [changes1, changes2]) - 一對一映射 備註 單一更新則捨棄回應表示以提升效能。 對於廣播或配對更新，該方法會委派給內部客戶端的批次更新邏輯。
upload_file	將檔案上傳到 Dataverse 的檔案欄位。 備註 大型檔案會自動分塊，以避免請求大小限制。 區塊模式可執行多次請求，並支援可重複上傳。

create

以資料表名稱建立一個或多個紀錄。

create(table_schema_name: str, records: Dict[str, Any] | List[Dict[str, Any]]) -> List[str]

參數

名稱	Description
table_schema_name 必要	str 表格的結構名稱（例如 "account"、 "contact"、 或 "new_MyTestTable"）。
records 必要	dict 或 list of dict 一本唱片字典，或是一串唱片字典清單。 每個字典都應該包含欄位結構名稱作為鍵。

傳回

類型	Description
list <xref:of> str	已建立的紀錄 GUID 列表。 回傳單一輸入的單一元素清單。

例外狀況

類型	Description
TypeError	如果 records 不是字典或 list[dict]，或是內部客戶端回傳意外型別。

範例

建立單一紀錄：

   client = DataverseClient(base_url, credential)
   ids = client.create("account", {"name": "Contoso"})
   print(f"Created: {ids[0]}")

建立多筆紀錄：

   records = [
       {"name": "Contoso"},
       {"name": "Fabrikam"}
   ]
   ids = client.create("account", records)
   print(f"Created {len(ids)} accounts")

create_columns

利用結構式映射在現有資料表上建立一個或多個欄位。

create_columns(table_schema_name: str, columns: Dict[str, Any]) -> List[str]

參數

名稱	Description
table_schema_name 必要	str 表格的結構名稱（例如 "new_MyTestTable"）。
columns 必要	dict mapping str to Any 欄位結構名稱（帶有自訂前綴值）對應到支援型別。 所有自訂欄位名稱必須包含自訂前綴值**（例如 "new_Notes"）。 原始型態包括 string、 int、 decimal、 floatdatetime、 bool和 。 Enum 子類別（IntEnum 優先）會產生局部選項集，並可透過 __labels__來指定本地化標籤。

傳回

類型	Description
list <xref:of> str	已建立欄位的結構名稱。

範例

在自訂表格上建立兩欄：

   created = client.create_columns(
       "new_MyTestTable",
       {
           "new_Scratch": "string",
           "new_Flags": "bool",
       },
   )
   print(created)  # ['new_Scratch', 'new_Flags']

create_table

建立一個簡單的自訂表格，並指定欄位。

create_table(table_schema_name: str, columns: Dict[str, Any], solution_unique_name: str | None = None, primary_column_schema_name: str | None = None) -> Dict[str, Any]

參數

名稱	Description
table_schema_name 必要	str 表格的結構名稱及自訂前綴值（例如 "new_MyTestTable"）。
columns 必要	dict mapping str to Any 字典將欄位名稱（帶有自訂前綴值）映射到其類型。 所有自訂欄位名稱必須包含自訂前綴值（例如 "new_Title"）。 支援類型： 原始型態："string"， "int"， "decimal"， "float""datetime"， ，"bool" Enum 子類別（IntEnum 優先）：建立本地選項集。 可透過類別屬性提供 __labels__ 可選的多語言標籤，該屬性定義於 Enum 子類別中： class ItemStatus(IntEnum): ACTIVE = 1 INACTIVE = 2 __labels__ = { 1033: {"Active": "Active", "Inactive": "Inactive"}, 1036: {"Active": "Actif", "Inactive": "Inactif"} }
solution_unique_name	str 選擇性解決方案，唯一名稱，應該擁有新資料表。 若省略，表格會以預設解法建立。 預設值: None
primary_column_schema_name	str 可選的主要名稱欄位結構名稱，並附有自訂前綴值（例如 "new_MyTestTable"）。 若未提供，則預設為 "{customization prefix value}_Name"。 預設值: None

傳回

類型	Description
dict	包含包含 table_schema_name、 entity_set_name、 table_logical_name、 metadata_id、 columns_created及 的表格元資料的字典。

例外狀況

類型	Description
MetadataError	如果資料表建立失敗或結構無效，

範例

建立一個以簡單欄位為表的表格：

   from enum import IntEnum

   class ItemStatus(IntEnum):
       ACTIVE = 1
       INACTIVE = 2

   columns = {
       "new_Title": "string",      # Note: includes 'new_' customization prefix value
       "new_Quantity": "int",
       "new_Price": "decimal",
       "new_Available": "bool",
       "new_Status": ItemStatus
   }

   result = client.create_table("new_MyTestTable", columns)
   print(f"Created table: {result['table_schema_name']}")
   print(f"Columns: {result['columns_created']}")

建立一個自訂主欄位名稱的表格：

   result = client.create_table(
       "new_Product",
       {"new_Price": "decimal"},
       primary_column_schema_name="new_ProductName"
   )

delete

透過 GUID 刪除一個或多個紀錄。

delete(table_schema_name: str, ids: str | List[str], use_bulk_delete: bool = True) -> str | None

參數

名稱	Description
table_schema_name 必要	str 表的結構名稱（例如 "account" 或 "new_MyTestTable"）。
ids 必要	str 或 list of str 刪除單一 GUID 字串或 GUID 字串清單。
use_bulk_delete	bool 當（預設）和 ids 為清單時True，執行 BulkDelete 動作並回傳其非同步工作識別碼。 每 False 筆紀錄依序刪除。 預設值: True

傳回

類型	Description
str, None	透過 BulkDelete 刪除多筆紀錄時的 BulkDelete 工作 ID;否則 None。

例外狀況

類型	Description
TypeError	如果 ids 不是 str 或 list[str]。
HttpError	如果底層的 Web API 刪除請求失敗，

範例

刪除一條紀錄：

   client.delete("account", account_id)

刪除多筆紀錄：

   job_id = client.delete("account", [id1, id2, id3])

delete_columns

從表格中刪除一個或多個欄位。

delete_columns(table_schema_name: str, columns: str | List[str]) -> List[str]

參數

名稱	Description
table_schema_name 必要	str 表格的結構名稱（例如 "new_MyTestTable"）。
columns 必要	str 或 list of str 欄位名稱或欄位名稱清單。 必須包含自訂前綴值（例如 "new_TestColumn"）。

傳回

類型	Description
list <xref:of> str	移除欄位的結構名稱。

範例

移除兩個依結構名稱設定的自訂欄位：

removed = client.delete_columns（「new_MyTestTable」、「[“new_Scratch”、“new_Flags”]，

） print（已移除） # ['new_Scratch'， 'new_Flags']

delete_table

以名稱刪除自訂資料表。

警告

此操作不可逆，會刪除表格中所有記錄

並定義了表格。 請小心使用。

delete_table(table_schema_name: str) -> None

參數

名稱	Description
table_schema_name 必要	str 表的結構名稱（例如 "new_MyTestTable" 或 "account"）。

例外狀況

類型	Description
MetadataError	如果表格不存在或刪除失敗。

範例

刪除自訂資料表：

   client.delete_table("new_MyTestTable")

flush_cache

清除快取的客戶端元資料或狀態。

flush_cache(kind) -> int

參數

名稱	Description
kind 必要	str 快取類型會沖洗。 目前支援的價值觀： "picklist"： 清除用於標籤轉整數轉換的選擇清單標籤快取 未來類型（例如 "entityset"、 "primaryid"）可以在不破壞此簽名的情況下加入。

傳回

類型	Description
int	移除的快取條目數量。

範例

清除選擇清單快取：

   removed = client.flush_cache("picklist")
   print(f"Cleared {removed} cached picklist entries")

get

可以依 ID 取得單一紀錄，或查詢多個紀錄。

提供時 record_id ，會回傳單一記錄字典。 當 record_id 為 None 時，回傳一個產生器，產生批量記錄。

get(table_schema_name: str, record_id: str | None = None, select: List[str] | None = None, filter: str | None = None, orderby: List[str] | None = None, top: int | None = None, expand: List[str] | None = None, page_size: int | None = None) -> Dict[str, Any] | Iterable[List[Dict[str, Any]]]

參數

名稱	Description
table_schema_name 必要	str 表的結構名稱（例如 "account" 或 "new_MyTestTable"）。
record_id	str 可選的 GUID 用來擷取特定紀錄。 若無，則查詢多筆紀錄。 預設值: None
select	list of str 可選的屬性邏輯名稱清單可供檢索。 欄位名稱不區分大小寫，且自動變小寫（例如 ["new_Title", "new_Amount"] 變成 "new_title,new_amount"）。 預設值: None
filter	str 可選的 OData 濾波串，例如 "name eq 'Contoso'""new_quantity gt 5"或 。 濾波器表達式中的欄位名稱必須使用精確的小寫邏輯名稱（例如 "new_quantity"，非 "new_Quantity"）。 過濾字串直接傳送到 Dataverse Web API，無需轉換。 預設值: None
orderby	list of str 可選的屬性排序清單，例如 ["name asc", "createdon desc"]。 欄位名稱會自動使用小寫。 預設值: None
top	int 可選地設定最多回傳紀錄數量。 預設值: None
expand	list of str 可擴充的導航屬性清單，例如 ["primarycontactid"]。 導航屬性名稱是區分大小寫的，必須與伺服器定義的名稱完全一致。 這些不會自動變換。 請參考實體元資料以確認正確的分類。 預設值: None
page_size	int 每頁可選擇記錄數量以供分頁。 預設值: None

傳回

類型	Description
dict, Iterable <xref:of> list <xref:of> dict	若 record_id 提供則使用單一記錄字典，否則則由產生器產生記錄字典清單（每頁一個清單）。

例外狀況

類型	Description
TypeError	如果 record_id 提供了，但沒有字串。

範例

取一張唱片：

   record = client.get("account", record_id=account_id, select=["name", "telephone1"])
   print(record["name"])

使用篩選查詢多筆紀錄（注意：篩選中需精確的邏輯名稱）：

   for batch in client.get(
       "account",
       filter="statecode eq 0 and name eq 'Contoso'",  # Must use exact logical names (lower-case)
       select=["name", "telephone1"]
   ):
       for account in batch:
           print(account["name"])

查詢時需展開導覽屬性（註：大小寫區分屬性名稱）：

   for batch in client.get(
       "account",
       select=["name"],
       expand=["primarycontactid"],  # Case-sensitive! Check metadata for exact name
       filter="statecode eq 0"
   ):
       for account in batch:
           print(f"{account['name']} - Contact: {account.get('primarycontactid', {}).get('fullname')}")

查詢與排序與分頁：

   for batch in client.get(
       "account",
       orderby=["createdon desc"],
       top=100,
       page_size=50
   ):
       print(f"Batch size: {len(batch)}")

get_table_info

如果有表格的基本元資料，請取得。

get_table_info(table_schema_name: str) -> Dict[str, Any] | None

參數

名稱	Description
table_schema_name 必要	str 表的結構名稱（例如 "new_MyTestTable" 或 "account"）。

傳回

類型	Description
dict, None	包含具有鍵 table_schema_name、 table_logical_name、 entity_set_name、 metadata_id和 的表格元資料的字典。 若找不到該表格，則回傳 無。

範例

擷取資料表元資料：

   info = client.get_table_info("new_MyTestTable")
   if info:
       print(f"Logical name: {info['table_logical_name']}")
       print(f"Entity set: {info['entity_set_name']}")

list_tables

列出 Dataverse 環境中所有自訂資料表。

list_tables() -> list[str]

傳回

類型	Description
list <xref:of> str	自訂表格名稱列表。

範例

列出所有自訂表格：

   tables = client.list_tables()
   for table in tables:
       print(table)

query_sql

使用 Dataverse Web API ?sql 功能執行唯讀 SQL 查詢。

SQL 查詢必須遵循支援的子集：一個 SELECT 陳述式，並可選地使用 WHERE、TOP（整數字面量）、ORDER BY（僅限欄位名稱）以及 FIN 後的簡單表格別名。

備註

SQL 支援僅限於唯讀查詢。 複雜的連接、子查詢以及某些 SQL 函式可能不被支援。 請參閱 Dataverse 文件以了解目前的功能集。

query_sql(sql: str)

參數

名稱	Description
sql 必要	str 支援 SQL SELECT 陳述式。

傳回

類型	Description
list <xref:of> dict	結果列字典列表。 若沒有相符的列，則回傳一個空清單。

例外狀況

類型	Description
SQLParseError	如果 SQL 查詢使用了不支援的語法，
HttpError	如果 Web API 回傳錯誤，

範例

基本 SQL 查詢：

   sql = "SELECT TOP 10 accountid, name FROM account WHERE name LIKE 'C%' ORDER BY name"
   results = client.query_sql(sql)
   for row in results:
       print(row["name"])

別名查詢：

   sql = "SELECT a.name, a.telephone1 FROM account AS a WHERE a.statecode = 0"
   results = client.query_sql(sql)

update

更新一個或多個紀錄。

此方法支援三種使用模式：

1. 單筆紀錄更新： update("account", "guid", {"name": "New Name"})

2. 廣播更新： update("account", [id1, id2], {"status": 1}) - 將相同的變更套用到所有 ID

3. 配對更新： update("account", [id1, id2], [changes1, changes2]) - 一對一映射

備註

單一更新則捨棄回應表示以提升效能。 對於廣播或配對更新，該方法會委派給內部客戶端的批次更新邏輯。

update(table_schema_name: str, ids: str | List[str], changes: Dict[str, Any] | List[Dict[str, Any]]) -> None

參數

名稱	Description
table_schema_name 必要	str 表的結構名稱（例如 "account" 或 "new_MyTestTable"）。
ids 必要	str 或 list of str 單一 GUID 字串或 GUID 字串清單來更新。
changes 必要	dict 或 list of dict 單播模式的變更字典，或配對模式的字典列表。 當 ids 是清單 且 changes 是單一字令時，相同的變更會廣播給所有記錄。 當兩者都是列表時，一對一映射時必須長度相等。

例外狀況

類型	Description
TypeError	If ids 不是 str 或 list[str]，或是 changes 類型與使用模式不符。

範例

單筆紀錄更新：

   client.update("account", account_id, {"telephone1": "555-0100"})

將相同變更廣播到多個唱片：

   client.update("account", [id1, id2, id3], {"statecode": 1})

更新多個不同值的紀錄：

   ids = [id1, id2]
   changes = [
       {"name": "Updated Name 1"},
       {"name": "Updated Name 2"}
   ]
   client.update("account", ids, changes)

upload_file

將檔案上傳到 Dataverse 的檔案欄位。

備註

大型檔案會自動分塊，以避免請求大小限制。 區塊模式可執行多次請求，並支援可重複上傳。

upload_file(table_schema_name: str, record_id: str, file_name_attribute: str, path: str, mode: str | None = None, mime_type: str | None = None, if_none_match: bool = True) -> None

參數

名稱	Description
table_schema_name 必要	str 表格的結構名稱，例如 "account""new_MyTestTable"或 。
record_id 必要	str 目標紀錄的 GUID。
file_name_attribute 必要	str 檔案欄位屬性的邏輯名稱。
path 必要	str 檔案的本地檔案系統路徑。 儲存的檔名會是這條路徑的基底名稱。
mode	str 上傳策略： "auto" （預設）、 "small"或 "chunk"。 自動模式會根據檔案大小選擇小上傳或分段上傳。 預設值: None
mime_type	str 與檔案一同儲存的明確 MIME 類型（例如 "application/pdf"）。 若未提供，則可從檔案副檔名推斷 MIME 類型。 預設值: None
if_none_match	bool 當 True（預設值）時，會 If-None-Match: null 發送標頭，只有當欄位目前為空時才會成功。 將 False 設為永遠覆 If-Match: *寫 。 僅用於小型和區塊模式。 預設值: True

例外狀況

類型	Description
HttpError	如果上傳失敗或檔案欄位不是空的，當 if_none_match=True
FileNotFoundError	如果指定的檔案路徑不存在。

範例

上傳PDF檔案：

   client.upload_file(
       table_schema_name="account",
       record_id=account_id,
       file_name_attribute="new_contract",
       path="/path/to/contract.pdf",
       mime_type="application/pdf"
   )

自動選擇模式上傳：

   client.upload_file(
       table_schema_name="email",
       record_id=email_id,
       file_name_attribute="new_attachment",
       path="/path/to/large_file.zip",
       mode="auto"
   )