Error Handling

The TitanRDM SDK uses a hierarchy of exceptions to communicate errors clearly. This guide covers exception types, common scenarios, and best practices for robust error handling.

---

Exception Hierarchy

TitanRDMError (base)
├── AuthenticationError    — OAuth authentication failures
├── APIError               — HTTP API request failures
├── ValidationError        — Input validation failures
├── ExportNotReadyError    — Export still in progress
├── ExportFailedError      — Export operation failed
└── UploadError            — S3 file upload failures

All exceptions inherit from TitanRDMError, so you can catch all SDK errors with a single handler if needed.

---

Exception Types

AuthenticationError

Raised when OAuth 2.0 authentication fails (invalid credentials, unreachable auth server).

from titan_rdm_sdk import TitanRDMClient, AuthenticationError

try:
    client = TitanRDMClient(url=URL, client_id=ID, client_secret=SECRET)
except AuthenticationError as e:
    print(f"Authentication failed: {e}")

Common causes: - Invalid client_id or client_secret - Incorrect instance URL - Network connectivity issues - Expired or revoked credentials

APIError

Raised when an API request returns a non-success HTTP status code.

from titan_rdm_sdk import APIError

try:
    upload = client.get_upload(branch_id=999999999)
except APIError as e:
    print(f"API error: {e}")
    print(f"HTTP status: {e.status_code}")
    print(f"Response: {e.response}")

Properties:

ValidationError

Raised when input parameters fail validation before making an API call.

from titan_rdm_sdk import ValidationError

try:
    client = TitanRDMClient(url="", client_id=ID, client_secret=SECRET)
except ValidationError as e:
    print(f"Invalid input: {e}")

Common causes: - Missing required parameters - Invalid pattern values - No branch/domain found matching the given name - Missing high_water_mark for incremental exports

ExportNotReadyError

Raised when you call download.receive() before the export is complete.

from titan_rdm_sdk import ExportNotReadyError

try:
    df = download.receive()
except ExportNotReadyError:
    print("Export still in progress — use wait_until_ready() first")

ExportFailedError

Raised when an export operation fails on the server side.

from titan_rdm_sdk import ExportFailedError

try:
    download.wait_until_ready()
    df = download.receive()
except ExportFailedError as e:
    print(f"Export failed: {e}")

UploadError

Raised when uploading a file to S3 fails (network issue, presigned URL expired).

from titan_rdm_sdk import UploadError

try:
    table_upload.send(df)
except UploadError as e:
    print(f"Upload failed: {e}")

---

Comprehensive Error Handling Example

from titan_rdm_sdk import (
    TitanRDMClient,
    AuthenticationError,
    APIError,
    ValidationError,
    ExportFailedError,
    ExportNotReadyError,
    UploadError,
)

try:
    # Authenticate
    client = TitanRDMClient(url=URL, client_id=ID, client_secret=SECRET)

    # Download
    download = client.get_download(
        branch_id=174,
        table_definition_key=50,
        pattern="full",
    )
    download.wait_until_ready(max_wait=120.0)
    df = download.receive()

except AuthenticationError as e:
    print(f"Authentication failed: {e}")
    # Action: Check credentials and URL

except ValidationError as e:
    print(f"Invalid parameters: {e}")
    # Action: Verify input values

except ExportFailedError as e:
    print(f"Export failed on server: {e}")
    # Action: Check TitanRDM logs, retry later

except TimeoutError as e:
    print(f"Export timed out: {e}")
    # Action: Increase max_wait or check server load

except APIError as e:
    print(f"API error (HTTP {e.status_code}): {e}")
    # Action: Check status code for specific handling

except UploadError as e:
    print(f"File upload failed: {e}")
    # Action: Retry the upload

---

Retry Strategies

Simple Retry

import time

def download_with_retry(client, branch_id, table_key, max_retries=3):
    for attempt in range(max_retries):
        try:
            download = client.get_download(
                branch_id=branch_id,
                table_definition_key=table_key,
                pattern="full",
            )
            download.wait_until_ready()
            return download.receive()
        except (APIError, ExportFailedError) as e:
            if attempt < max_retries - 1:
                wait = 2 ** attempt  # Exponential backoff
                print(f"Attempt {attempt + 1} failed: {e}. Retrying in {wait}s...")
                time.sleep(wait)
            else:
                raise

Token Refresh

The SDK handles token refresh automatically. If you encounter authentication errors mid-session, the token may have been revoked server-side:

try:
    result = client.get_branches()
except AuthenticationError:
    # Re-create the client to force re-authentication
    client = TitanRDMClient(url=URL, client_id=ID, client_secret=SECRET)

---

Best Practices

1. Catch specific exceptions — Handle each exception type with appropriate recovery logic
2. Use wait_until_ready() — Always wait for exports before calling receive()
3. Set reasonable timeouts — Use max_wait to prevent indefinite blocking
4. Log correlation codes — Include correlation_code in operations for traceability
5. Don't swallow errors — Log or re-raise exceptions to avoid silent failures

---

Next Steps

• Uploading Data — Upload workflows with error handling
• Downloading Data — Download workflows with polling
• SDK Client Methods — All available methods