titan-rdm-sdk 0.1.5

Python SDK for TitanRdm Upload and Download APIs

TitanRDM SDK for Python

Python SDK for TitanRDM Upload and Download APIs.

Installation

pip install titan-rdm-sdk

For development:

pip install titan-rdm-sdk[dev]

Quick Start

Initialize Client

from titan_rdm_sdk import TitanRDMClient

client = TitanRDMClient(
    url="https://<tenant_subdomain>.titanrdm.com",
    client_id="your_client_id",
    client_secret="your_client_secret"
)

Download Data (Full Export)

download = client.get_download(
    branch_id=174,
    table_definition_key=50,
    pattern='full',
    correlation_code='daily-export-2025-01-22'
)

download.wait_until_ready(poll_interval=2.0, max_wait=300.0)
df = download.receive()
print(f"Downloaded {len(df)} rows")

Download Data (Incremental Export)

download = client.get_download(
    branch_id=174,
    table_definition_key=50,
    pattern='incremental',
    high_water_mark="2025-01-21T00:00:00Z",
    correlation_code='incremental-sync'
)

download.wait_until_ready()
df = download.receive()

Upload Data

import pandas as pd

# Create upload batch
upload = client.get_upload(
    branch_id=174,
    description='Customer data update',
    correlation_code='batch-001'
)

# Prepare data
df = pd.DataFrame({
    'customer_id': [1, 2, 3],
    'name': ['Alice', 'Bob', 'Charlie'],
    'email': ['[e-mail]', '[e-mail]', '[e-mail]']
})

# Upload to specific table
table_upload = upload.get_table_upload(
    table_definition_key=456,
    import_mapping_key=123,
    pattern='full'
)
table_upload.send(df)

# Complete the import batch
upload.complete()

Error Handling

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

try:
    client = TitanRDMClient(url, client_id, client_secret)
    download = client.get_download(branch_id=174, table_definition_key=50)
    download.wait_until_ready()
    df = download.receive()

except AuthenticationError as e:
    print(f"Authentication failed: {e}")

except ValidationError as e:
    print(f"Invalid input: {e}")

except ExportFailedError as e:
    print(f"Export failed: {e}")

except APIError as e:
    print(f"API error (HTTP {e.status_code}): {e}")

Development

Setup

# Clone the repository
git clone https://github.com/dimodelo/titan-rdm-sdk-python.git
cd titan-rdm-sdk-python

# Install in development mode
pip install -e ".[dev]"

Running Tests

# Run all tests
pytest

# Run with verbose output
pytest -v

# Run with coverage report
pytest --cov=titan_rdm_sdk --cov-report=term-missing

# Run specific test file
pytest tests/test_client.py

# Run specific test class or function
pytest tests/test_client.py::TestClientInitialization
pytest tests/test_client.py::TestClientInitialization::test_successful_initialization

Building the Package

pip install build
python -m build

This creates distribution files in the dist/ directory that can be uploaded to PyPI.

Testing in Databricks

For comprehensive testing of the SDK in a Databricks environment, use the provided system test notebook:

Databricks System Tests

The databricks/databricks_system_tests/ directory contains a ready-to-use Databricks notebook that tests all major SDK functionality:

Features:

• Authentication testing
• Full upload/download workflows
• Incremental operations
• Error handling scenarios
• Integration with both DBFS and Unity Catalog volumes

Setup:

1. Upload the notebook to your Databricks workspace
2. Configure secrets for API credentials
3. Upload test data to DBFS or Unity Catalog volumes
4. Run the tests using configurable widgets

Quick Start:

# In Databricks notebook
%pip install titan-rdm-sdk

# Import the system test functions
%run /path/to/databricks/databricks_system_tests.py

# Configure test parameters
dbutils.widgets.text("branch_id", "174")
dbutils.widgets.text("table_definition_key", "50")
dbutils.widgets.text("test_csv_path", "/test_data/customers.csv")

# Run all tests
run_all_tests()

For detailed setup instructions, see the databricks/databricks_README.md file.

License

MIT