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})

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

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

Upsert (作成と更新)

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

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

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

from PowerPlatform.Dataverse.models.upsert 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 を使用します。

import pandas as pd
from PowerPlatform.Dataverse.models.filters 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",
)

バッチ操作

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")

バッチ内のテーブル メタデータと 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()

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

関連情報

• テーブルと列をカスタマイズする
• データのクエリ
• SDK for Python のコード例
• SDK for Python README

こちらも参照ください

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