Python SDK を使用して Dataverse データを操作する

この記事では、SDK を使用して Dataverse のデータとメタデータを操作するコード例を示します。 続行する前に、「作業の 開始」を必ずお読みください。

基本操作

アカウント テーブルで動作するコード例を次に示します。

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

# Replace <myorg> with the name of a valid environment.
base_url = "https://<myorg>.crm.dynamics.com"
client = DataverseClient(base_url=base_url, credential=InteractiveBrowserCredential())

# Create a record
account_id = client.records.create("account", {"name": "Contoso Ltd"})

# Read a record
account = client.records.retrieve("account", account_id)
print(account["name"])

# Read with expand fetches a related record in the same HTTP request
account = client.records.retrieve(
    "account", account_id,
    select=["name"],
    expand=["primarycontactid"],
)
contact = (account.get("primarycontactid") or {})
print(contact.get("fullname"))

# Update a record
client.records.update("account", account_id, {"telephone1": "555-0199"})

# Delete a record
client.records.delete("account", account_id)

コンテキスト マネージャー

コンテキスト マネージャーは、自動クリーンアップと HTTP 接続プールを処理します。 コンテキスト マネージャーを利用するには、次の構文を使用します。

with DataverseClient("https://<myorg>.crm.dynamics.com", credential) as client:

次の作業コードは、コンテキスト マネージャーの使用方法を示しています。

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

# Connect to Dataverse
credential = InteractiveBrowserCredential()

with DataverseClient("https://<myorg>.crm.dynamics.com", credential) as client:

    # Create a contact
    contact_id = client.records.create("contact", {"firstname": "John", "lastname": "Doe"})

    # Read the contact back
    contact = client.records.retrieve("contact", contact_id, select=["firstname", "lastname"])
    print(f"Created: {contact['firstname']} {contact['lastname']}")

    # Clean up
    client.records.delete("contact", contact_id)

# Session closed, caches cleared automatically

一括操作

一括操作を実行する例をいくつか次に示します。

# Bulk create
payloads = [
    {"name": "Company A"},
    {"name": "Company B"},
    {"name": "Company C"}
]
ids = client.records.create("account", payloads)

# Bulk update (broadcast same change to all)
client.records.update("account", ids, {"industry": "Technology"})

# Bulk delete
client.records.delete("account", ids, use_bulk_delete=True)

次の例では、複数のアカウントを作成します。 create(logical_name, payloads)にペイロードの一覧を渡して、コレクション バインドMicrosoft.Dynamics.CRM.CreateMultipleアクションを呼び出します。 このメソッドは、作成されたレコード ID の list[str] を返します。

# Bulk create accounts (returns list of GUIDs)
payloads = [
    {"name": "Contoso"},
    {"name": "Fabrikam"},
    {"name": "Northwind"},
]
ids = client.records.create("account", payloads)
assert isinstance(ids, list) and all(isinstance(x, str) for x in ids)
print({"created_ids": ids})

一括操作の詳細については、以下を参照してください。

• update は、セマンティクスの一貫性を保つために、単一レコード呼び出しと一括呼び出しの両方の None を返します。
• ブロードキャストとレコードごとの比較は、 changes パラメーターがディクショナリかリストかによって決まります。
• 主キー属性は、 UpdateMultiple アクション ターゲットを構築するときに自動的に挿入されます。
• ペイロードが @odata.typeを省略すると、SDK によって自動的にスタンプされます (キャッシュされた論理名の参照)。
• 応答には ID のみが含まれます。SDK はそれらの GUID 文字列を返します。
• create は、1 つのディクショナリを渡すときに新しいレコードの GUID を str として返し、ペイロードの一覧を渡すときに list[str] を返します。
• @odata.typeのメタデータ検索は、エンティティ セット (メモリ内キャッシュ) ごとに 1 回実行されます。

Upsert (作成と更新)

一般的なデータ アクセス シーケンスは、最初にテーブル行が存在するかどうかを確認することです。 行が存在する場合は、更新します。 そうでない場合は、行を作成します。 Upsert 操作の 1 つの API 呼び出しを使用することで、このシーケンスをより効率的にすることができます。

詳細については、「 Upsert を使用してレコードを作成または更新する」を参照してください。

Important

テーブルには、 alternate_keyで使用される列の代替キーが Dataverse で構成されている必要があります。 Power Apps Maker ポータルまたは Dataverse API 呼び出しを使用して、テーブルのメタデータで代替キーを定義します。 代替キーが構成されていない場合、Dataverse は 400 エラーで upsert 要求を拒否します。

client.records.upsert()を使用して、代替キーによって識別されるレコードを作成または更新します。 キーが既存のレコードと一致すると、メソッドはレコードを更新します。 それ以外の場合は、レコードが作成されます。 1 つの項目で PATCH 要求が使用され、複数の項目で UpsertMultiple 一括アクションが使用されます。

from PowerPlatform.Dataverse.models import UpsertItem

# Upsert a single record
client.records.upsert("account", [
    UpsertItem(
        alternate_key={"accountnumber": "ACC-001"},
        record={"name": "Contoso Ltd", "telephone1": "555-0100"},
    )
])

# Upsert multiple records (uses UpsertMultiple bulk action)
client.records.upsert("account", [
    UpsertItem(
        alternate_key={"accountnumber": "ACC-001"},
        record={"name": "Contoso Ltd"},
    ),
    UpsertItem(
        alternate_key={"accountnumber": "ACC-002"},
        record={"name": "Fabrikam Inc"},
    ),
])

# Composite alternate key (multiple columns identify the record)
client.records.upsert("account", [
    UpsertItem(
        alternate_key={"accountnumber": "ACC-001", "address1_postalcode": "98052"},
        record={"name": "Contoso Ltd"},
    )
])

# Plain dict syntax (no import needed)
client.records.upsert("account", [
    {
        "alternate_key": {"accountnumber": "ACC-001"},
        "record": {"name": "Contoso Ltd"},
    }
])

データフレーム (DataFrames)

SDK は、 client.dataframe 名前空間を介して、すべての CRUD 操作に pandas ラッパーを提供します。 これらのラッパーは、入力と出力に pandas の DataFrame API と Series API を使用します。

Note

client.dataframe.get() は非推奨とされます。 次のセクションに示す GA パターンを使用します。

import pandas as pd
from PowerPlatform.Dataverse.models import col

# Query records as a single DataFrame (GA builder pattern)
df = (client.query.builder("account")
      .select("name", "telephone1")
      .where(col("statecode") == 0)
      .execute()
      .to_dataframe())
print(f"Found {len(df)} accounts")

# Limit results with top for large tables
df = client.query.builder("account").select("name").top(100).execute().to_dataframe()

# Create records from a DataFrame (returns a Series of GUIDs)
new_accounts = pd.DataFrame([
    {"name": "Contoso", "telephone1": "555-0100"},
    {"name": "Fabrikam", "telephone1": "555-0200"},
])
new_accounts["accountid"] = client.dataframe.create("account", new_accounts)

# Update records from a DataFrame (id_column identifies the GUID column)
new_accounts["telephone1"] = ["555-0199", "555-0299"]
client.dataframe.update("account", new_accounts, id_column="accountid")

# Clear a field by setting clear_nulls=True (by default, NaN/None fields are skipped)
df = pd.DataFrame([{"accountid": new_accounts["accountid"].iloc[0], "websiteurl": None}])
client.dataframe.update("account", df, id_column="accountid", clear_nulls=True)

# Delete records by passing a Series of GUIDs
client.dataframe.delete("account", new_accounts["accountid"])

# SQL query directly to DataFrame (supports JOINs, aggregates, GROUP BY)
df = client.dataframe.sql(
    "SELECT a.name, COUNT(c.contactid) as contacts "
    "FROM account a "
    "JOIN contact c ON a.accountid = c.parentcustomerid "
    "GROUP BY a.name"
)

Dataverse にファイルをアップロードする

次の例では、document.pdfという名前のファイルを、アカウント レコードの という名前の new_Documentにアップロードする方法を示します。 sdk for Pythonは、128 MB を超えるファイルのファイル チャンクを自動的に処理します。

# Upload a file to a record
client.files.upload(
    "account",
    account_id,
    "new_Document",
    "/path/to/document.pdf",
)

Important

ファイル列が存在しない場合は、SDK によって自動的に作成されます。 列の作成はメタデータの変更です。テーブルをカスタマイズするには特権が必要であり、新しい列は環境のスキーマの永続的な部分になります。 運用環境では、自動作成に依存するのではなく、事前にファイル列を作成します。

upload メソッドは、コンテンツ タイプと転送戦略を制御するための省略可能なパラメーターを受け入れます。

# Specify a MIME type (defaults to application/octet-stream)
client.files.upload(
    "account", account_id, "new_Contract", "/path/to/contract.pdf",
    mime_type="application/pdf",
)

# Force chunked transfer regardless of size
client.files.upload(
    "email", email_id, "new_Attachment", "/path/to/large_file.zip",
    mode="chunk",
)

# Overwrite an existing file (uploads fail by default when the column already holds a file)
client.files.upload(
    "account", account_id, "new_Document", "/path/to/updated_contract.pdf",
    if_none_match=False,
)

mode パラメーターは、SDK がファイルを転送する方法を制御します。

Mode	Behavior
"auto" (既定値)	128 MB 未満のファイルの "small" 、または 128 MB 以上のファイルの "chunk" を選択します。
"small"	1 つの要求でファイル全体を送信します。 ファイルが 128 MB を超える場合に ValueError を発生させます。
"chunk"	ファイルをセグメント単位でストリーミングします。 大きなファイルに使用するか、サイズに関係なくチャンク転送を強制する場合に使用します。

Note

Dataverse ファイル サイズの制限は、ファイルのアップロードに適用されます (既定では、ファイルあたり 128 MB)。 mode="small"を使用して大きなファイルをアップロードすると、ValueErrorが発生します。 mode="chunk"または既定のmode="auto"を使用します。 非同期クライアントは、同じ files.upload メソッドを公開します。 awaitで呼び出します。

バッチ操作

client.batchを使用して、1 つの HTTP 要求で複数の操作を送信します。 バッチ名前空間は、 client.records、 client.tables、および client.queryを反映します。

# Build a batch request and add operations
batch = client.batch.new()
batch.records.create("account", {"name": "Contoso"})
batch.records.create("account", [{"name": "Fabrikam"}, {"name": "Woodgrove"}])
batch.records.update("account", account_id, {"telephone1": "555-0100"})
batch.records.delete("account", old_id)
batch.records.retrieve("account", account_id, select=["name"], expand=["primarycontactid"])  # single record with expand
batch.records.list(                                                # multi-record, single page
    "account",
    filter="statecode eq 0",
    select=["name"],
    orderby=["name asc"],
    top=50,
)

result = batch.execute()
for item in result.responses:
    if item.is_success:
        print(f"[OK] {item.status_code} entity_id={item.entity_id}")
    else:
        print(f"[ERR] {item.status_code}: {item.error_message}")

トランザクション変更セット

変更セット内のすべての操作は、まとめて成功するか、まとめてロールバックされます。

batch = client.batch.new()
with batch.changeset() as cs:
    lead_ref = cs.records.create("lead", {"firstname": "Ada"})
    contact_ref = cs.records.create("contact", {"firstname": "Ada"})
    cs.records.create("account", {
        "name": "Babbage & Co.",
        "originatingleadid@odata.bind": lead_ref,
        "primarycontactid@odata.bind": contact_ref,
    })
result = batch.execute()
print(f"Created {len(result.entity_ids)} records atomically")

変更セット内 cs.records.create() は、 "$1"などのコンテンツ ID 参照文字列を返します。 @odata.bind フィールドの値 (前述) として、または後のcs.records.update()またはcs.records.delete()呼び出しのrecord_id引数として、サーバーの応答を待たずに、同じ変更セットで使用します。 変更セット内では、読み取り操作 (retrieve、 list) は許可されません。

バッチ内のテーブル メタデータと SQL クエリ

batch = client.batch.new()
batch.tables.create("new_Product", {"new_Price": "decimal", "new_InStock": "bool"})
batch.tables.add_columns("new_Product", {"new_Rating": "int"})
batch.tables.get("new_Product")
batch.query.sql("SELECT TOP 5 name FROM account")

result = batch.execute()

エラーが発生した場合に続行する

失敗した場合でも、すべての操作を試行します。

result = batch.execute(continue_on_error=True)
print(f"Succeeded: {len(result.succeeded)}, Failed: {len(result.failed)}")
for item in result.failed:
    print(f"[ERR] {item.status_code}: {item.error_message}")

DataFrame の統合

pandas DataFrames をバッチに直接フィードします。

import pandas as pd

batch = client.batch.new()

# Create records from a DataFrame
df = pd.DataFrame([{"name": "Contoso"}, {"name": "Fabrikam"}])
batch.dataframe.create("account", df)

# Update records from a DataFrame
updates = pd.DataFrame([
    {"accountid": id1, "telephone1": "555-0100"},
    {"accountid": id2, "telephone1": "555-0200"},
])
batch.dataframe.update("account", updates, id_column="accountid")

# Delete records from a Series
batch.dataframe.delete("account", pd.Series([id1, id2]))

result = batch.execute()

バッチ応答と制限を処理する

batch.execute() は、BatchResult を返します。 その responses リストには、送信順に HTTP 操作ごとに 1 つのエントリが含まれています。 succeededとfailedリストは、responsesのサブセットです。 has_errors プロパティは、操作が失敗したときにTrueされます。 entity_idsリストは、成功した単一レコードの作成および更新のOData-EntityIdヘッダーからGUIDを収集します。

各応答エントリは、 status_code、 is_success、 entity_id、 data (取得、一覧表示、およびクエリ操作の解析された本文)、 error_message、および error_codeを公開します。 異なる操作によって、異なる図形が返されます。

• 単一レコードの作成 — ステータス 204、entity_id が設定される。
• 一括作成またはアップサート — ステータス 200、ID は item.data["Ids"] 内（OData-EntityId ヘッダーなし）。
• SQL クエリまたはリスト - 状態 200、 item.data["value"]内の行。
• 削除またはメタデータの書き込み - 状態 204、本文データなし。

バッチあたりの HTTP 操作の最大数は 1,000 です。 この制限を超えると、クライアントは要求を送信する前に ValidationError を発生させます。 一部の操作は複数の要求に拡張されます。 batch.tables.add_columns() と batch.tables.remove_columns() は、各列ごとに 1 つの要求に解決されるため、各列は独自の応答エントリを生成します。

完全なバッチの例については、examples/advanced/batch.py を参照してください。

関連情報

• テーブルと列をカスタマイズする
• データのクエリ
• エラーを処理し、HTTP 診断を有効にする
• SDK for Python のコード例
• Python向けSDK README

こちらも参照ください

• 作業の開始
• Dataverse のクイック ガイド