19.7.1 Import Data Entities using Import Job API

This topic documents the Python Import Job API for importing data entities.

The following example connects to the active deployment and starts an import job directly from Python. Replace the vault-password, connection, and schema placeholders before running it.

By default this method returns session name only for backward compatibility. Set return_details=True to get full job details.

from datatransforms.workbench import DataTransformsWorkbench, WorkbenchConfig

connect_params = WorkbenchConfig.get_workbench_config(
    "<your-workbench-password-from-vault>"
)
workbench = DataTransformsWorkbench()
workbench.connect_workbench(connect_params)

result = workbench.import_data_entities(
    connection_name="<connection-name>",
    schema_name="<schema-name>",
    import_options={
        # Option names depend on connector template
        # "SOME_OPTION": "value"
    },
    wait_for_status=True,
    poll_interval=5,
    max_wait=900,
    progress_callback=print,
    return_details=True
)

print("Session Name :", result["session_name"])
print("Job ID       :", result["job_id"])
print("Final Status :", result["final_status"])

For a BICC connection, leave schema_name empty and provide the selected offerings through LIST_OF_OFFERINGS as shown in Import Options for BICC.

Import Options for BICC

For BICC connections:

  • schema_name is not required for import behavior.
  • Offerings must be provided through option LIST_OF_OFFERINGS.
  • Offerings value must be comma-separated.

You can read offerings catalog first:

from datatransforms.clients import DataEntityClient

de_client = DataEntityClient(workbench)
offerings = de_client.get_bicc_offerings_for_connection("Demo BICC")
print(offerings)

result = workbench.import_data_entities(
    connection_name="Demo BICC",
    schema_name="",   # ignored for BICC
    import_options={
        "LIST_OF_OFFERINGS": "Financial,Sales,Human Resources"
    },
    wait_for_status=True,
    return_details=True
)

print("Final Status:", result["final_status"])