Advanced
Error Handling

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:

ChannelWhenDefault SDK behavior
HTTP status4xx/5xx from the transport layerRaises HttpError or a DealCloud* exception
Row body ErrorsHTTP 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

GoalAPIReturn typeRaises on row "Errors"?
Bulk / ETL, inspect manuallydefault write calllist (mixed)No
Bulk with transport resilienceerror_handling=COLLECTBatchResultNo (use .row_errors)
Typed inspection, no exceptionoutput="write_result"RowsWriteResultNo (use .has_row_errors)
Strict job / fail pipelineraise_on_row_errors=TrueN/AYes → DealCloudValidationError
Org-wide strict defaultDealCloudConfig(raiseOnRowErrors=True)inheritsYes 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 TypeDescriptionCommon Causes
DealCloudApiErrorBase SDK API errorAPI returned error details
DealCloudValidationErrorValidation errorInvalid field values, missing required fields
DealCloudRateLimitErrorRate limited (429)Too many requests; use retry_after
DealCloudAuthErrorAuth/authorization401, 403; check credentials and scope
HttpErrorHTTP layer (intapp-rest-client)Non-success status, transport errors
ValueErrorInvalid inputMissing required params, bad format
TimeoutExceptionRequest timeoutLarge operations, network issues
ValidationErrorPydantic errorsTyped 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

CodeMeaningResolution
400Bad RequestCheck request parameters
401UnauthorizedVerify credentials
403ForbiddenCheck API permissions
404Not FoundVerify object/entry exists
429Rate LimitedReduce request frequency
500Server ErrorRetry with backoff
503UnavailableRetry 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:
        raise

ErrorHandling 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 config

Retry 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