Error Handling
Comprehensive guide to handling errors in the DealCloud SDK. For a full list of exception types and attributes, see the Exceptions Reference.
Transport failures from intapp-rest-client surface as HttpError (status_code, response_body, optional retry_after). The SDK may wrap or translate some responses into DealCloud* exceptions below.
Two error channels
Rows write operations (insert_data, update_data, upsert_data, delete_data) can fail in two independent ways:
| Channel | When | Default SDK behavior |
|---|---|---|
| HTTP status | 4xx/5xx from the transport layer | Raises HttpError or a DealCloud* exception |
Row body Errors | HTTP 200 with per-row "Errors" (e.g. code 5009 uniqueness) | Returns the full API row list; logs each error row; does not raise |
ErrorHandling (FAIL_FAST, COLLECT, LOG) controls parallel transport failures (e.g. one batch HTTP 502). It does not replace row-level "Errors" handling—you still inspect the return value or use the helpers below.
Applies to Rows writes (insert_data, update_data, upsert_data, delete_data) and write_cells.
Row error handling decision matrix
| Goal | API | Return type | Raises on row "Errors"? |
|---|---|---|---|
| Bulk / ETL, inspect manually | default write call | list (mixed) | No |
| Bulk with transport resilience | error_handling=COLLECT | BatchResult | No (use .row_errors) |
| Typed inspection, no exception | output="write_result" | RowsWriteResult | No (use .has_row_errors) |
| Strict job / fail pipeline | raise_on_row_errors=True | N/A | Yes → DealCloudValidationError |
| Org-wide strict default | DealCloudConfig(raiseOnRowErrors=True) | inherits | Yes unless per-call override |
RowsWriteResult is also returned by write_cells with output="write_result". Single-entry Cells helpers (backlink_dms_document, create_entry_with_store_requests) support raise_on_row_errors only—use write_cells or Rows APIs for COLLECT / write_result.
raise_on_row_errors=True (optional on write methods) raises DealCloudValidationError when any returned row contains "Errors". The exception includes row_errors (full row dicts with EntryId) and errors (flattened field errors).
from dealcloud_sdk import DealCloud, split_row_results, DealCloudValidationError
# Default: caller inspects the list (HTTP 200, mixed ok + error rows)
result = dc.update_data("Company", records)
ok_rows, row_errors = split_row_results(result)
# Structured wrapper (FAIL_FAST only)
write_result = dc.update_data("Company", records, output="write_result")
if write_result.has_row_errors:
for row in write_result.row_errors:
print(row["EntryId"], row["Errors"])
# Strict: fail the job on any row error
try:
dc.insert_data("Company", records, raise_on_row_errors=True)
except DealCloudValidationError as e:
for row in e.row_errors:
print(row["EntryId"], row["Errors"])Error Types
| Error Type | Description | Common Causes |
|---|---|---|
DealCloudApiError | Base SDK API error | API returned error details |
DealCloudValidationError | Validation error | Invalid field values, missing required fields |
DealCloudRateLimitError | Rate limited (429) | Too many requests; use retry_after |
DealCloudAuthError | Auth/authorization | 401, 403; check credentials and scope |
HttpError | HTTP layer (intapp-rest-client) | Non-success status, transport errors |
ValueError | Invalid input | Missing required params, bad format |
TimeoutException | Request timeout | Large operations, network issues |
ValidationError | Pydantic errors | Typed data validation failures |
Basic Error Handling
from dealcloud_sdk import DealCloud, DealCloudConfig, DealCloudApiError, DealCloudRateLimitError
from intapp_rest_client.exceptions import HttpError
config = DealCloudConfig(
siteUrl="yoursite.dealcloud.com",
clientId=12345,
clientSecret="your-secret",
)
dc = DealCloud.from_config_object(config)
try:
data = dc.read_data("Company", output="pandas")
except DealCloudRateLimitError as e:
print(f"Rate limited; retry after {getattr(e, 'retry_after', 60)}s")
except DealCloudApiError as e:
print(f"API error: {e.message} (status={e.status_code})")
except HttpError as e:
print(f"HTTP error: {e.status_code}")
except ValueError as e:
print(f"Invalid input: {e}")
except Exception as e:
print(f"Unexpected error: {e}")HTTP Status Codes
| Code | Meaning | Resolution |
|---|---|---|
| 400 | Bad Request | Check request parameters |
| 401 | Unauthorized | Verify credentials |
| 403 | Forbidden | Check API permissions |
| 404 | Not Found | Verify object/entry exists |
| 429 | Rate Limited | Reduce request frequency |
| 500 | Server Error | Retry with backoff |
| 503 | Unavailable | Retry later |
Handling Specific Status Codes
import httpx
try:
data = dc.read_data("InvalidObject", output="list")
except httpx.HTTPStatusError as e:
if e.response.status_code == 404:
print("Object not found")
elif e.response.status_code == 401:
print("Authentication failed - check credentials")
elif e.response.status_code == 429:
print("Rate limited - slow down requests")
else:
raiseErrorHandling Enum
Control batch operation error behavior:
from dealcloud_sdk import DealCloud, DealCloudConfig, ErrorHandling
config = DealCloudConfig(
siteUrl="yoursite.dealcloud.com",
clientId=12345,
clientSecret="your-secret",
)
dc = DealCloud.from_config_object(config)# Stop on first transport error (default)
try:
result = dc.insert_data(
"Company",
records,
error_handling=ErrorHandling.FAIL_FAST
)
except Exception as e:
print(f"Operation failed: {e}")Validation Errors
For typed data operations:
from pydantic import BaseModel, ValidationError
class Company(BaseModel):
EntryId: int
CompanyName: str
Revenue: float # Required to be float
try:
companies = dc.typed_read_data(Company, object_id="Company")
except ValidationError as e:
print(f"Data validation failed: {e}")
# Details on each error
for error in e.errors():
print(f" Field: {error['loc']}")
print(f" Error: {error['msg']}")Skip Invalid Records
# Don't fail on validation errors
companies = dc.typed_read_data(
Company,
object_id="Company",
skip_validation_errors=True # Log warnings, skip bad records
)Timeout Handling
import httpx
try:
data = dc.read_data("LargeObject", output="pandas")
except httpx.TimeoutException:
print("Request timed out - try streaming or increase timeout")
# Option 1: Use streaming
for record in dc.read_data_streaming("LargeObject"):
process(record)
# Option 2: Increase timeout in configRetry Logic
The SDK has built-in retry for transient errors. For custom retry:
import time
from functools import wraps
def retry(max_attempts=3, delay=1, backoff=2):
"""Decorator for retry with exponential backoff."""
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
attempts = 0
current_delay = delay
while attempts < max_attempts:
try:
return func(*args, **kwargs)
except (httpx.HTTPStatusError, httpx.TimeoutException) as e:
attempts += 1
if attempts >= max_attempts:
raise
print(f"Attempt {attempts} failed, retrying in {current_delay}s...")
time.sleep(current_delay)
current_delay *= backoff
return wrapper
return decorator
@retry(max_attempts=3, delay=1, backoff=2)
def robust_read(dc, object_id):
return dc.read_data(object_id, output="list")Common Error Patterns
Missing Required Output Parameter
# Error: ValueError - output is required
try:
data = dc.read_data("Company") # Missing output=
except ValueError as e:
print(f"Migration required: {e}")
# Fix: Add output parameter
data = dc.read_data("Company", output="pandas")Invalid Object Name
try:
data = dc.read_data("NonExistentObject", output="list")
except httpx.HTTPStatusError as e:
if e.response.status_code == 404:
# List valid objects
objects = dc.get_objects()
print("Valid objects:", [o.apiName for o in objects])Invalid Field Name
try:
data = dc.read_data(
"Company",
output="list",
fields=["InvalidFieldName"]
)
except Exception as e:
# Get valid fields
fields = dc.get_fields("Company")
print("Valid fields:", [f.apiName for f in fields])Error Logging
import logging
# Configure logging
logging.basicConfig(level=logging.WARNING)
logger = logging.getLogger("dealcloud_sdk")
logger.setLevel(logging.DEBUG)
# Errors are now logged with full details
try:
dc.read_data("Company", output="list")
except Exception as e:
logger.exception("Failed to read data")Custom Exception Handler
class DealCloudErrorHandler:
"""Centralized error handling."""
def __init__(self, dc):
self.dc = dc
def safe_read(self, object_id, **kwargs):
try:
return self.dc.read_data(object_id, **kwargs)
except httpx.HTTPStatusError as e:
self._handle_http_error(e)
except ValueError as e:
self._handle_value_error(e)
except Exception as e:
self._handle_unknown_error(e)
return None
def _handle_http_error(self, e):
status = e.response.status_code
if status == 401:
raise AuthenticationError("Invalid credentials")
elif status == 403:
raise PermissionError("Access denied")
elif status == 429:
raise RateLimitError("Too many requests")
else:
raise
def _handle_value_error(self, e):
logging.error(f"Invalid input: {e}")
raise
def _handle_unknown_error(self, e):
logging.exception("Unexpected error")
raise
# Usage
handler = DealCloudErrorHandler(dc)
data = handler.safe_read("Company", output="list")Client config default (raiseOnRowErrors)
from dealcloud_sdk import DealCloud, DealCloudConfig, DealCloudValidationError
config = DealCloudConfig(
siteUrl="yoursite.dealcloud.com",
clientId=12345,
clientSecret="your-secret",
raiseOnRowErrors=True,
)
dc = DealCloud.from_config_object(config)
# Inherits config default — raises on any row Errors
try:
dc.insert_data("Company", records)
except DealCloudValidationError as e:
print(len(e.row_errors))
# Per-call override for a bulk job on the same client
result = dc.insert_data("Company", records, raise_on_row_errors=False)Related
- Error Enrichment — opt-in
fieldId→ schema metadata and formatted messages (not automatic) - Configuration - Retry settings
- Logging - Debug logging
- Performance - Avoiding errors