Based on official Microsoft documentation for Azure SDK error handling patterns and Dataverse SDK specifics.
1. DataverseError Class Overview
The Dataverse SDK for Python provides a structured exception hierarchy for robust error handling.
DataverseError Constructor
from PowerPlatform.Dataverse.core.errors import DataverseError
DataverseError(
message: str, # Human-readable error message
code: str, # Error category (e.g., "validation_error", "http_error")
subcode: str | None = None, # Optional specific error identifier
status_code: int | None = None, # HTTP status code (if applicable)
details: Dict[str, Any] | None = None, # Additional diagnostic information
source: str | None = None, # Error source: "client" or "server"
is_transient: bool = False # Whether error may succeed on retry
)
Key Properties
try:
client.get("account", record_id="invalid-id")
except DataverseError as e:
print(f"Message: {e.message}") # Human-readable message
print(f"Code: {e.code}") # Error category
print(f"Subcode: {e.subcode}") # Specific error type
print(f"Status Code: {e.status_code}") # HTTP status (401, 403, 429, etc.)
print(f"Source: {e.source}") # "client" or "server"
print(f"Is Transient: {e.is_transient}") # Can retry?
print(f"Details: {e.details}") # Additional context
# Convert to dictionary for logging
error_dict = e.to_dict()
2. Common Error Scenarios
Authentication Errors (401)
Cause: Invalid credentials, expired tokens, or misconfigured settings.
from PowerPlatform.Dataverse.client import DataverseClient
from PowerPlatform.Dataverse.core.errors import DataverseError
from azure.identity import InteractiveBrowserCredential
try:
# Bad credentials or expired token
credential = InteractiveBrowserCredential()
client = DataverseClient(
base_url="https://invalid-org.crm.dynamics.com",
credential=credential
)
records = client.get("account")
except DataverseError as e:
if e.status_code == 401:
print("Authentication failed. Check credentials and token expiration.")
print(f"Details: {e.message}")
# Don't retry - fix credentials first
else:
raise
Authorization Errors (403)
Cause: User lacks permissions for the requested operation.
try:
# User doesn't have permission to read contacts
records = client.get("contact")
except DataverseError as e:
if e.status_code == 403:
print("Access denied. User lacks required permissions.")
print(f"Request ID for support: {e.details.get('request_id')}")
# Escalate to administrator
else:
raise
Resource Not Found (404)
Cause: Record, table, or resource doesn't exist.
try:
# Record doesn't exist
record = client.get("account", record_id="00000000-0000-0000-0000-000000000000")
except DataverseError as e:
if e.status_code == 404:
print("Resource not found. Using default data.")
record = {"name": "Unknown", "id": None}
else:
raise
Rate Limiting (429)
Cause: Too many requests exceeding service protection limits.
Note: The SDK has minimal built-in retry support. Handle transient consistency issues manually.
import time
def create_with_retry(client, table_name, payload, max_retries=3):
"""Create record with retry logic for rate limiting."""
for attempt in range(max_retries):
try:
result = client.create(table_name, payload)
return result
except DataverseError as e:
if e.status_code == 429 and e.is_transient:
wait_time = 2 ** attempt # Exponential backoff
print(f"Rate limited. Retrying in {wait_time}s...")
time.sleep(wait_time)
else:
raise
raise Exception(f"Failed after {max_retries} retries")
Server Errors (500, 502, 503, 504)
Cause: Temporary service issues or infrastructure problems.
try:
result = client.create("account", {"name": "Acme"})
except DataverseError as e:
if 500 <= e.status_code < 600:
print(f"Server error ({e.status_code}). Service may be temporarily unavailable.")
# Implement retry logic with exponential backoff
else:
raise
Validation Errors (400)
Cause: Invalid request format, missing required fields, or business rule violations.
try:
# Missing required field or invalid data
client.create("account", {"telephone1": "not-a-phone-number"})
except DataverseError as e:
if e.status_code == 400:
print(f"Validation error: {e.message}")
if e.details:
print(f"Details: {e.details}")
# Log validation issues for debugging
else:
raise
3. Error Handling Best Practices
Use Specific Exception Handling
Always catch specific exceptions before general ones:
from PowerPlatform.Dataverse.core.errors import DataverseError
from azure.core.exceptions import AzureError
try:
records = client.get("account", filter="statecode eq 0", top=100)
except DataverseError as e:
# Handle Dataverse-specific errors
if e.status_code == 401:
print("Re-authenticate required")
elif e.status_code == 404:
print("Resource not found")
elif e.is_transient:
print("Transient error - may retry")
else:
print(f"Operation failed: {e.message}")
except AzureError as e:
# Handle Azure SDK errors (network, auth, etc.)
print(f"Azure error: {e}")
except Exception as e:
# Catch-all for unexpected errors
print(f"Unexpected error: {e}")
Implement Smart Retry Logic
Don't retry on:
- 401 Unauthorized (authentication failures)
- 403 Forbidden (authorization failures)
- 400 Bad Request (client errors)
- 404 Not Found (unless resource should eventually appear)
Consider retrying on:
- 408 Request Timeout
- 429 Too Many Requests (with exponential backoff)
- 500 Internal Server Error
- 502 Bad Gateway
- 503 Service Unavailable
- 504 Gateway Timeout
def should_retry(error: DataverseError) -> bool:
"""Determine if operation should be retried."""
if not error.is_transient:
return False
retryable_codes = {408, 429, 500, 502, 503, 504}
return error.status_code in retryable_codes
def call_with_exponential_backoff(func, *args, max_attempts=3, **kwargs):
"""Call function with exponential backoff retry."""
for attempt in range(max_attempts):
try:
return func(*args, **kwargs)
except DataverseError as e:
if should_retry(e) and attempt < max_attempts - 1:
wait_time = 2 ** attempt # 1s, 2s, 4s...
print(f"Attempt {attempt + 1} failed. Retrying in {wait_time}s...")
time.sleep(wait_time)
else:
raise
Extract Meaningful Error Information
import json
from datetime import datetime
def log_error_for_support(error: DataverseError):
"""Log error with diagnostic information."""
error_info = {
"timestamp": datetime.utcnow().isoformat(),
"error_type": type(error).__name__,
"message": error.message,
"code": error.code,
"subcode": error.subcode,
"status_code": error.status_code,
"source": error.source,
"is_transient": error.is_transient,
"details": error.details
}
print(json.dumps(error_info, indent=2))
# Save to log file or send to monitoring service
return error_info
Handle Bulk Operations Gracefully
def bulk_create_with_error_tracking(client, table_name, payloads):
"""Create multiple records, tracking which succeed/fail."""
results = {
"succeeded": [],
"failed": []
}
for idx, payload in enumerate(payloads):
try:
record_ids = client.create(table_name, payload)
results["succeeded"].append({
"payload": payload,
"ids": record_ids
})
except DataverseError as e:
results["failed"].append({
"index": idx,
"payload": payload,
"error": {
"message": e.message,
"code": e.code,
"status": e.status_code
}
})
return results
4. Enable Diagnostic Logging
Configure Logging
import logging
import sys
# Set up root logger
logging.basicConfig(
level=logging.DEBUG,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
handlers=[
logging.FileHandler('dataverse_sdk.log'),
logging.StreamHandler(sys.stdout)
]
)
# Configure specific loggers
logging.getLogger('azure').setLevel(logging.DEBUG)
logging.getLogger('PowerPlatform').setLevel(logging.DEBUG)
# HTTP logging (careful with sensitive data)
logging.getLogger('azure.core.pipeline.policies.http_logging_policy').setLevel(logging.DEBUG)
Enable SDK-Level Logging
from PowerPlatform.Dataverse.client import DataverseClient
from PowerPlatform.Dataverse.core.config import DataverseConfig
from azure.identity import InteractiveBrowserCredential
cfg = DataverseConfig()
cfg.logging_enable = True # Enable detailed logging
client = DataverseClient(
base_url="https://myorg.crm.dynamics.com",
credential=InteractiveBrowserCredential(),
config=cfg
)
# Now SDK will log detailed HTTP requests/responses
records = client.get("account", top=10)
Parse Error Responses
import json
try:
client.create("account", invalid_payload)
except DataverseError as e:
# Extract structured error details
if e.details and isinstance(e.details, dict):
error_code = e.details.get('error', {}).get('code')
error_message = e.details.get('error', {}).get('message')
print(f"Error Code: {error_code}")
print(f"Error Message: {error_message}")
# Some errors include nested details
if 'error' in e.details and 'details' in e.details['error']:
for detail in e.details['error']['details']:
print(f" - {detail.get('code')}: {detail.get('message')}")
5. Dataverse-Specific Error Handling
Handle OData Query Errors
try:
# Invalid OData filter
records = client.get(
"account",
filter="invalid_column eq 0"
)
except DataverseError as e:
if "invalid column" in e.message.lower():
print("Check OData column names and syntax")
else:
print(f"Query error: {e.message}")
Handle File Upload Errors
try:
client.upload_file(
table_name="account",
record_id=record_id,
column_name="document_column",
file_path="large_file.pdf"
)
except DataverseError as e:
if e.status_code == 413:
print("File too large. Use chunked upload mode.")
elif e.status_code == 400:
print("Invalid column or file format.")
else:
raise
Handle Table Metadata Operations
try:
# Create custom table
table_def = {
"SchemaName": "new_CustomTable",
"DisplayName": "Custom Table"
}
client.create("EntityMetadata", table_def)
except DataverseError as e:
if "already exists" in e.message:
print("Table already exists")
elif "permission" in e.message.lower():
print("Insufficient permissions to create tables")
else:
raise
6. Monitoring and Alerting
Wrap Client Calls with Monitoring
from functools import wraps
import time
def monitor_operation(operation_name):
"""Decorator to monitor SDK operations."""
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
start_time = time.time()
try:
result = func(*args, **kwargs)
duration = time.time() - start_time
print(f"✓ {operation_name} completed in {duration:.2f}s")
return result
except DataverseError as e:
duration = time.time() - start_time
print(f"✗ {operation_name} failed after {duration:.2f}s")
print(f" Error: {e.code} ({e.status_code}): {e.message}")
raise
return wrapper
return decorator
@monitor_operation("Fetch Accounts")
def get_accounts(client):
return client.get("account", top=100)
# Usage
try:
accounts = get_accounts(client)
except DataverseError:
print("Operation failed - check logs for details")
7. Common Troubleshooting Checklist
| Issue | Diagnosis | Solution |
|---|---|---|
| 401 Unauthorized | Expired token or bad credentials | Re-authenticate with valid credentials |
| 403 Forbidden | User lacks permissions | Request access from administrator |
| 404 Not Found | Record/table doesn't exist | Verify schema name and record ID |
| 429 Rate Limited | Too many requests | Implement exponential backoff retry |
| 500+ Server Error | Service issue | Retry with exponential backoff; check status page |
| 400 Bad Request | Invalid request format | Check OData syntax, field names, required fields |
| Network timeout | Connection issues | Check network, increase timeout in DataverseConfig |
| InvalidOperationException | Plugin/workflow error | Check plugin logs in Dataverse |
8. Logging Best Practices
import logging
import json
from datetime import datetime
class DataverseErrorHandler:
"""Centralized error handling and logging."""
def __init__(self, log_file="dataverse_errors.log"):
self.logger = logging.getLogger("DataverseSDK")
handler = logging.FileHandler(log_file)
formatter = logging.Formatter(
'%(asctime)s - %(levelname)s - %(message)s'
)
handler.setFormatter(formatter)
self.logger.addHandler(handler)
self.logger.setLevel(logging.ERROR)
def log_error(self, error: DataverseError, context: str = ""):
"""Log error with context for debugging."""
error_record = {
"timestamp": datetime.utcnow().isoformat(),
"context": context,
"error": error.to_dict()
}
self.logger.error(json.dumps(error_record, indent=2))
def is_retryable(self, error: DataverseError) -> bool:
"""Check if error should be retried."""
return error.is_transient and error.status_code in {408, 429, 500, 502, 503, 504}
# Usage
error_handler = DataverseErrorHandler()
try:
client.create("account", payload)
except DataverseError as e:
error_handler.log_error(e, "create_account_batch_1")
if error_handler.is_retryable(e):
print("Will retry this operation")
else:
print("Operation failed permanently")