Exceptions

from dealcloud_sdk import DealCloudApiError
 
try:
    dc.insert_data("Company", bad_data)
except DealCloudApiError as e:
    print(f"Message: {e.message}")
    print(f"Status: {e.status_code}")  # Optional
    print(f"Errors: {e.errors}")       # List of error details

from dealcloud_sdk import DealCloudValidationError
 
try:
    dc.update_data("Company", records, raise_on_row_errors=True)
except DealCloudValidationError as e:
    print(f"Validation failed: {e.message}")
    for row in e.row_errors:
        print(f"  EntryId {row['EntryId']}: {row['Errors']}")
    for err in e.errors:
        print(f"  field={err.get('field')} code={err.get('code')}")

from dealcloud_sdk import DealCloudRateLimitError
import time
 
try:
    dc.read_data("Company", output="list")
except DealCloudRateLimitError as e:
    wait = getattr(e, "retry_after", None) or 60
    print(f"Rate limited; retry after {wait}s")
    time.sleep(wait)
    # Retry request

from dealcloud_sdk import DealCloudAuthError
 
try:
    dc.get_users()
except DealCloudAuthError as e:
    print(f"Auth failed: {e.message}")
    # Check credentials, scope, permissions

try:
    dc.read_data("Company")  # Missing required 'output' parameter
except ValueError as e:
    print(f"Invalid input: {e}")

import httpx
 
try:
    dc.read_data("InvalidObject", output="list")
except httpx.HTTPStatusError as e:
    print(f"Status: {e.response.status_code}")
    print(f"URL: {e.request.url}")
    print(f"Body: {e.response.text}")

import httpx
 
try:
    dc.read_data("LargeObject", output="pandas")
except httpx.TimeoutException:
    print("Request timed out")
    # Use streaming or increase timeout

import httpx
 
try:
    dc = DealCloud(site_url="invalid.domain.com", ...)
    dc.get_objects()
except httpx.ConnectError:
    print("Could not connect to server")

from pydantic import ValidationError
 
try:
    companies = dc.typed_read_data(Company, object_id="Company")
except ValidationError as e:
    print(f"Validation failed: {e}")
    for error in e.errors():
        print(f"  {error['loc']}: {error['msg']}")

try:
    dc.insert_data("Company", invalid_data)
except httpx.HTTPStatusError as e:
    # Parse error details
    import json
    try:
        error_details = e.response.json()
        print(f"Error code: {error_details.get('errorCode')}")
        print(f"Message: {error_details.get('message')}")
    except json.JSONDecodeError:
        print(f"Raw error: {e.response.text}")

import httpx
from pydantic import ValidationError
 
def safe_operation(func):
    """Decorator for comprehensive error handling."""
    def wrapper(*args, **kwargs):
        try:
            return func(*args, **kwargs)
        except ValueError as e:
            print(f"Invalid input: {e}")
            return None
        except httpx.HTTPStatusError as e:
            status = e.response.status_code
            if status == 401:
                print("Authentication failed")
            elif status == 404:
                print("Resource not found")
            elif status == 429:
                print("Rate limited - try again later")
            else:
                print(f"API error: {status}")
            return None
        except httpx.TimeoutException:
            print("Request timed out")
            return None
        except ValidationError as e:
            print(f"Data validation failed: {e}")
            return None
        except Exception as e:
            print(f"Unexpected error: {e}")
            raise
    return wrapper
 
@safe_operation
def read_companies(dc):
    return dc.read_data("Company", output="list")

import time
import httpx
 
def retry_on_error(func, max_retries=3, backoff=2):
    """Retry function with exponential backoff."""
    for attempt in range(max_retries):
        try:
            return func()
        except httpx.HTTPStatusError as e:
            if e.response.status_code in [429, 500, 503]:
                delay = backoff ** attempt
                print(f"Retrying in {delay}s...")
                time.sleep(delay)
            else:
                raise
        except httpx.TimeoutException:
            delay = backoff ** attempt
            print(f"Timeout, retrying in {delay}s...")
            time.sleep(delay)
    
    raise Exception(f"Failed after {max_retries} retries")
 
# Usage
data = retry_on_error(lambda: dc.read_data("Company", output="list"))

from dealcloud_sdk import DealCloud, ErrorHandling
 
def batch_with_errors(dc, object_id, records):
    """Process batch collecting all errors."""
    
    result = dc.insert_data(
        object_id,
        records,
        error_handling=ErrorHandling.COLLECT
    )
    
    success = list(result.results)
    errors = [
        {"data": row, "errors": row["Errors"]}
        for row in result.row_errors
    ]
    
    return {
        "success": len(success),
        "failed": len(errors) + len(result.errors),
        "errors": errors
    }

import logging
 
logger = logging.getLogger("dealcloud_app")
 
def logged_operation(dc, object_id):
    try:
        return dc.read_data(object_id, output="list")
    except httpx.HTTPStatusError as e:
        logger.error(
            f"API error reading {object_id}",
            extra={
                "status_code": e.response.status_code,
                "url": str(e.request.url),
                "response": e.response.text[:500]
            }
        )
        raise
    except Exception as e:
        logger.exception(f"Unexpected error reading {object_id}")
        raise