Introduction to Data Transfer (Export)¶
Data transfer in Elevate Data enables you to seamlessly move prepared datasets or imported data tables to Price-It and Underwrite-It. This functionality ensures that the data you have imported, curated and transformed within Elevate Data can be leveraged across the entire Earnix application ecosystem without manual intervention. The transfer process is designed to be secure, efficient, and flexible to meet diverse business needs.
The data transfer process typically includes the following stages:
- Update Type Mapping – Align source data formats with the specific requirements of the target system.
- Configuration of Target – Specify where Elevate Data should transfer Data to: Destination, location, naming convention.
- Selection of Dataset/Data table – Choose the specific version of the dataset or data table to be transferred.
- Follow-up and Completion – Review the status and confirm successful transfer.
Install the SDK¶
!pip install --upgrade earnix-elevate
Set your Elevate server and credentials¶
In this example we use environment variables to authenticate to Elevate, but you can also inject your credentials using Python arguments to each Service's client, like in the commented example.
import os
os.environ.setdefault("E2_SERVER", "YOUR_SERVER")
os.environ.setdefault("E2_CLIENT_ID", "YOUR_CLIENT_ID")
os.environ.setdefault("E2_SECRET_KEY", "YOUR_SECRET_KEY")
# Alternatively:
# ConnectionService(
# server="YOUR_SERVER",
# client_id="YOUR_CLIENT_ID",
# secret_key="YOUR_SECRET_KEY"
# )
Imports and demo preparation¶
from datetime import datetime
from time import sleep
from earnix_elevate import (
ConnectionService,
CreateExportRequest,
CreatePriceItDataTargetRequest,
DataTargetService,
ExportService,
UpdateConnectionTypeMappingRequest,
)
from earnix_elevate.clients.imx import JobStatus
DEMO_SUFFIX = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
Update Price-It Connection Type Mapping¶
Data type mapping is the process of aligning source data formats with the specific requirements of a target system to prevent errors and ensure data integrity.
The following code demonstrates how to update these mappings to match export data types with Price-It table columns. Once updated, these definitions apply to all future exports using that connection. The process involves fetching a (Price-It) connection by its ID, preparing a request with the new mapping configurations, and executing an update call.
import os
CONNECTION_ID = int(os.environ.get("EXPORT_CONNECTION_ID", "36602")) # use your connection system id
connection_service = ConnectionService()
connection = connection_service.get_connection(CONNECTION_ID)
update_request = UpdateConnectionTypeMappingRequest(
id=connection.id,
lockVersion=connection.lock_version,
typeMapping='{"AGE": "integer", "NAME": "text"}',
typeMappingName="Simple Mapping",
)
connection_service.update_connection_type_mapping(connection.id, update_request)
Create Price-It Data Target¶
A target is a pre-configured destination for delivering data during the transfer process, defined per connection type. For a Price-It target, it includes parameters such as project path, table path, and table naming conventions. Once created, the target entity is reusable across multiple transfers, ensuring consistency and reducing setup effort.
To create a Price-It target, first create a data target request. Next, initialize a data target service and invoke its creation method with the request. You need to provide an identifier of an existing Price-It connection. The table name may include dynamic fields, that would be replaced with a concrete string during the export. The fields are: source_name, source_version_number and timestamp.
data_target_name = os.environ.get("EXPORT_DATA_TARGET_NAME", "PriceIt Dim Data Target Test " + DEMO_SUFFIX)
project_path = os.environ.get("EXPORT_PROJECT_PATH", "YOUR_PROJECT_PATH")
data_target_request = CreatePriceItDataTargetRequest(
name=data_target_name,
description="PriceIt Dim Desc",
connectionId=connection.id,
type="price_it",
projectPath=project_path,
tablePath="Data Tables\\",
tableName="table_{{source_name}}_{{timestamp}}",
)
data_target_service = DataTargetService()
data_target = data_target_service.create_data_target(data_target_request)
# Alternatively, if data target already exists, you can use its ID to fetch it
# data_target = data_target_service.get_data_target(DATA_TARGET_ID)
print(f"\n{data_target=}")
Export Data Table / Dataset¶
After the data target has been successfully created, the export operation can be performed. To do so, create an export request and use the export service to execute it.
The export request supports two source types: 'datatable' and 'dataset', representing data tables and datasets, respectively. The source ID should correspond to the identifier of the data table or dataset to be exported.
DATA_TABLE_ID = int(os.environ.get("EXPORT_DATA_TABLE_ID", "25852")) # use your data table system id
export_datatable_request = CreateExportRequest(
targetId=data_target.id,
sourceType="datatable",
sourceId=DATA_TABLE_ID,
sourceVersion=1,
useSample=False,
validateToRelease=True,
castBooleanToInteger=False,
)
export_service = ExportService()
export_response = export_service.export(export_datatable_request)
print(f"\n{export_response=}")
Follow-up on the export status of the exported data¶
export_id = export_response.export_id
MAX_ATTEMPTS = 180 # Wait max 15 minutes (180 * 5 sec)
WAIT_INTERVAL = 5 # seconds
status_code = None
for attempt in range(MAX_ATTEMPTS):
try:
export_response = export_service.get_export(export_id)
status_code = export_response.execution.status if export_response.execution else None
if status_code == JobStatus.SUCCEEDED:
print("\nThe export was completed successfully\n")
break
elif status_code == JobStatus.FAILED:
print("\nThe export operation failed\n")
break
except Exception as ex:
print(f"Exception during status polling: {ex}")
sleep(WAIT_INTERVAL)
print(f"Waiting Export:{export_id}\n")
else:
raise RuntimeError(f"Timed out waiting for export {export_id} to complete.")
Delete the Data Target¶
data_target_service.delete_data_target(data_target_id=data_target.id)