Trabajar con datos de Dataverse mediante el SDK de Python

En este artículo se muestra código de ejemplo que usa el SDK para trabajar con datos y metadatos de Dataverse. Antes de continuar, asegúrese de leer Introducción.

Operaciones Básico

Este es un ejemplo de código que funciona en la tabla de cuentas.

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)

Administrador de contextos

El Administrador de contextos controla la limpieza automática y la agrupación de conexiones HTTP. Use la sintaxis siguiente para aprovechar el Administrador de contextos.

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

El código de trabajo siguiente muestra cómo usar el Administrador de contextos.

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

Operaciones masivas

Estos son un par de ejemplos que realizan operaciones masivas.

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

En el ejemplo siguiente se crean varias cuentas. Pase una lista de cargas a create(logical_name, payloads) para invocar la acción enlazada a la colección de Microsoft.Dynamics.CRM.CreateMultiple. El método devuelve list[str] los identificadores de registro creados.

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

Para obtener más información sobre las operaciones masivas:

• update devuelve None tanto para las llamadas de un solo registro como para las llamadas masivas, para mantener una semántica coherente.
• La elección entre transmisión y procesamiento por registro se determina según si el parámetro changes es un diccionario o una lista.
• El atributo de clave principal se inserta automáticamente al construir destinos de acción UpdateMultiple .
• Si alguna carga omite @odata.type, el SDK lo marca automáticamente (búsqueda de nombres lógicos almacenados en caché).
• La respuesta solo incluye identificadores: el SDK devuelve esas cadenas GUID.
• create devuelve el GUID del nuevo registro como str al pasar un solo diccionario, y como list[str] al pasar una lista de cargas útiles.
• La búsqueda @odata.type de metadatos se realiza una vez por conjunto de entidades (en memoria caché).

Upsert (crear y actualizar)

Una secuencia de acceso a datos común es comprobar primero si existe una fila de tabla. Si la fila existe, actualícela. De lo contrario, cree la fila. Puede hacer que esta secuencia sea más eficaz mediante una sola llamada API de la operación Upsert.

Para obtener más información, consulte Uso de Upsert para crear o actualizar un registro.

Use client.records.upsert() para crear o actualizar registros identificados por claves alternativas. Cuando la clave coincide con un registro existente, el método actualiza el registro. De lo contrario, crea el registro. Un solo elemento usa una petición PATCH, mientras que varios elementos usan la UpsertMultiple acción en bloque.

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"},
    }
])

Marcos de datos

El SDK proporciona contenedores pandas para todas las operaciones CRUD a través del espacio de nombres client.dataframe. Estos contenedores usan las API de panda, DataFrame y Series para la entrada y salida.

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

Carga de archivos en Dataverse

En el ejemplo siguiente se muestra cómo cargar un archivo denominado document.pdf en la columna Archivo denominado new_Document de un registro de cuenta. El SDK para Python controla automáticamente la fragmentación de archivos para los archivos de más de 128 MB.

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

El upload método acepta parámetros opcionales para controlar el tipo de contenido y la estrategia de transferencia.

# 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,
)

El mode parámetro controla cómo transfiere el SDK el archivo.

Operaciones por lotes

Use client.batch para enviar varias operaciones en una solicitud HTTP. El espacio de nombres por lotes refleja client.records, client.tables y 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}")

Conjunto de cambios transaccionales

Todas las operaciones de un conjunto de cambios se completan o se revierten todas juntas.

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

Dentro de un conjunto de cambios, cs.records.create() devuelve una cadena de referencia content-ID como "$1". Úselo en el mismo conjunto de cambios, sin esperar una respuesta de servidor, ya sea como el valor de un @odata.bind campo (como se mostró anteriormente) o como argumento record_id para una llamada posterior cs.records.update() o cs.records.delete() . No se permiten operaciones de lectura (retrieve, list) dentro de un conjunto de cambios.

Metadatos de tabla y consultas SQL en un lote

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

Continuar ante un error

Intente todas las operaciones incluso cuando se produzca un error.

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

Integración de DataFrame

Alimentar DataFrames de Pandas directamente en un lote.

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

Gestionar respuestas por lotes y límites

batch.execute() devuelve BatchResult. Su responses lista contiene una entrada por operación HTTP en orden de envío. Las succeeded listas y failed son subconjuntos de responses. La has_errors propiedad es True cuando se produce un error en cualquier operación. La lista entity_ids recopila los GUID del encabezado OData-EntityId de las operaciones correctas de creación y actualización de un solo registro.

Cada entrada de respuesta expone status_code, is_success, entity_id, data (cuerpo analizado para las operaciones de recuperación, lista y consulta), error_messagey error_code. Las distintas operaciones devuelven formas diferentes:

• Creación de un único registro : estado 204, entity_id rellenado.
• Creación o actualización masiva — estado 200, IDs en item.data["Ids"] (sin encabezado OData-EntityId).
• Consulta o lista SQL : estado 200, filas en item.data["value"].
• Eliminación o escritura de metadatos — estado 204, sin datos en el cuerpo.

El número máximo de operaciones HTTP por lote es de 1000. Si supera este límite, el cliente genera un ValidationError antes de enviar la solicitud. Algunas operaciones se expanden a más de una solicitud: batch.tables.add_columns() y batch.tables.remove_columns() cada una resuelve en una solicitud por columna, por lo que cada columna genera su propia entrada de respuesta.

Para obtener un ejemplo de lote completo, consulte examples/advanced/batch.py.

Información relacionada

• Personalización de tablas y columnas
• Consultar datos
• Control de errores y habilitación de diagnósticos HTTP
• Ejemplos de código del SDK para Python
• SDK para Python README

Consulte también

• Introducción
• Guía rápida de Dataverse