Cloud Dataverse
binaryrain_helper_cloud_dataverse
is a python package that aims to simplify and help with connecting to and using Microsoft Dataverse. It handles common operations like retrieving, creating, updating, and deleting records in a Dataverse environment. With the help of sessions it maintains a consistent connection to the Dataverse API, ensuring efficient and reliable data operations without the need for repetitive code when pagination is required.
Installation
Section titled “Installation”To install the package you can use your favorite python package manager:
pip install binaryrain-helper-cloud-dataverse
uv add binaryrain-helper-cloud-dataverse
Key Functions
Section titled “Key Functions”DataverseAPIHandler()
Section titled “DataverseAPIHandler()”an API handler that makes interaction with Dataverse tables easier.
from binaryrain_helper_cloud_dataverse.dataverse import DataverseAPIHandler
api_handler = DataverseAPIHandler( base_url="<your-service-uri>", # something like: xxxxx.crm4.dynamics.com access_token="<your_access_token")
Parameters:
Section titled “Parameters:”base_url
:str
| The url of the dataverse serviceaccess_token
:str
| The access token for your dataverse service
Retrieve records from a Dataverse table with automatic pagination handling.
# Get all contactscontacts = api_handler.get("contacts")
# Get contacts with specific query parametersfiltered_contacts = api_handler.get( "contacts", params={ "$select": "fullname,emailaddress1", "$filter": "statecode eq 0" })
# Get with custom timeoutaccounts = api_handler.get("accounts", timeout=120)
Parameters:
Section titled “Parameters:”endpoint
:str
| The API endpoint (e.g., ‘contacts’, ‘accounts’)timeout
:int
| Request timeout in seconds. Default is 60 secondsparams
:dict | None
| Query parameters to include in the request (optional)**kwargs
| Additional arguments to pass to requests.get()
Returns:
Section titled “Returns:”list[dict]
| A list of records retrieved from the Dataverse API
Raises:
Section titled “Raises:”requests.RequestException
| If the request fails
post()
Section titled “post()”Create a new record in a Dataverse table.
# Create a new contactcontact_data = { "firstname": "John", "lastname": "Doe", "emailaddress1": "john.doe@example.com"}contact_id = api_handler.post("contacts", contact_data)
# Create a new account with custom timeoutaccount_data = { "name": "Acme Corporation", "websiteurl": "https://acme.com"}account_id = api_handler.post("accounts", account_data, timeout=90)
Parameters:
Section titled “Parameters:”endpoint
:str
| The API endpoint (e.g., ‘contacts’, ‘accounts’)data
:dict
| The JSON data to send in the request bodytimeout
:int
| Request timeout in seconds. Default is 60 seconds**kwargs
| Additional arguments to pass to requests.post()
Returns:
Section titled “Returns:”str | None
| The created record ID or None if creation failed
Raises:
Section titled “Raises:”requests.RequestException
| If the request failsValueError
| If data is None
delete()
Section titled “delete()”Delete a record from a Dataverse table.
# Delete a contact by IDcontact_id = "12345678-1234-1234-1234-123456789012"success = api_handler.delete("contacts", contact_id)
if success: print("Contact deleted successfully")else: print("Failed to delete contact")
# Delete with custom timeoutaccount_id = "87654321-4321-4321-4321-210987654321"success = api_handler.delete("accounts", account_id, timeout=30)
Parameters:
Section titled “Parameters:”endpoint
:str
| The API endpoint (e.g., ‘contacts’, ‘accounts’)record_id
:str
| The GUID of the record to deletetimeout
:int
| Request timeout in seconds. Default is 60 seconds**kwargs
| Additional arguments to pass to requests.delete()
Returns:
Section titled “Returns:”bool
| True if deletion was successful, False otherwise
Raises:
Section titled “Raises:”requests.RequestException
| If the request fails