Create DataLoad From Schema & BICC Tables

When the BICC schema and participating tables are known, use source with the BICC connection and schema name. For BICC Data Loads, use incremental_merge and incremental_merge_delete table actions.

This approach is best suited for small, curated table lists. BICC offerings typically expose many tables, so maintaining one Python table call per table is not the recommended workflow for broad offering-level loads. For larger selections, prefer exporting the participating tables to CSV and attaching the reviewed CSV back to the Data Load.

For incremental_merge, pass incremental_column=None and merge_keys=None. This matches the working BICC examples under demo_examples/dataloads.

from datatransforms.dataload import DataLoad

dl = DataLoad("demo_bicc_dl", project_name="BICC_DEMO_PROJECT")

(
    dl.source("DEMO_BICC.FscmTopModel")
    .incremental_merge(
        "CrmAnalyticsAM.ActivitiesAM.Activity",
        incremental_column=None,
        merge_keys=None)
    .incremental_merge(
        "CrmAnalyticsAM.ActivitiesAM.ActivityAssignee",
        incremental_column=None,
        merge_keys=None)
    .incremental_merge_delete(
        "CrmAnalyticsAM.ActivitiesAM.ActivityContact")
    .target("tgt.TGT_USER")
)

created = dl.create_dataload()

Create DataLoad Using BICC Offerings And Live Tables

Use bicc_source when the BICC schema should be resolved by Data Transforms from selected offerings. Create the Data Load with offerings first, export the BICC live tables from that Data Load, then attach the reviewed table-selection CSV or action-list folder before saving the final table selection.

A BICC Data Load must have participating tables for the final load definition. Those tables can come from explicit table calls, a manifest CSV, or an action-list folder.

How To Export Offerings To CSV

Export offerings when developers want to review and version-control the selected offering list. The exported CSV has one column named offering.

bicc_client = workbench.get_bicc_client()

offerings_file = bicc_client.export_matching_offerings_as_csv(
    connection_name="DEMO_BICC",
    output_file="bicc_offerings.csv")

To export only matching offerings, pass matching. The matcher supports regular expressions, and unescaped * or % are treated as wildcards.

offerings_file = bicc_client.export_matching_offerings_as_csv(
    connection_name="DEMO_BICC",
    matching="Financial|%Service%",
    output_file="bicc_offerings.csv")

For large BICC models, export live tables from an existing Data Load and keep the table-selection artifact under version control. Use Data Load names in developer scripts; avoid hardcoding generated identifiers.

How To Export BICC Live Tables And Actions

Manifest mode creates one CSV with all live tables returned by the Data Load. Use action_rules when table actions can be classified by table name during export.

action_rules = {
    DataLoad.ACTION_INCREMENTAL_MERGE: "ABC%|DEF*",
    DataLoad.ACTION_INCREMENTAL_WITH_DELETES: {
        "include": "FGH*",
        "exclude": "*ADE*",
    },
    DataLoad.ACTION_DO_NOT_LOAD: "TEMP%|TEST%",
}

table_selection_file = dl.export_bicc_live_tables(
    action_rules=action_rules,
    output_file="bicc_table_selection.csv")

If output_file is omitted, the SDK derives the file name from the Data Load name.

Unmatched tables remain unassigned. If a table matches more than one action rule, the SDK raises DataTransformsException so the rules can be fixed before the CSV is committed.

Attach Participating Tables To The Data Load

After the manifest CSV is reviewed, attach it to the Data Load with prepare_data_load_tables. The CSV is read immediately and re-read when create_dataload is called, so the checked-in CSV is the source of truth.

dl.prepare_data_load_tables(csv_file="bicc_table_selection.csv")
created = dl.create_dataload()

For BICC table-selection manifests, only table_name and target_action are used. INCREMENTAL_MERGE becomes an incremental merge table, and INCREMENTAL_WITH_DELETES becomes INCREMENTAL_MERGE_DELETE in the Data Load payload. Blank and DO_NOT_LOAD rows are ignored.

Export Participating Tables As Action Lists

Action-list mode creates one CSV per action. This is easier to review when the table count is large.

selection_dir = dl.export_bicc_live_tables(
    mode=DataLoad.EXPORT_LIVE_TABLES_ACTION,
    action_rules=action_rules)

If output_dir is omitted, the SDK derives a directory from the Data Load name:

demo_bicc_from_offerings/
  unassigned.csv
  incremental_merge.csv
  incremental_with_deletes.csv
  do_not_load.csv

Each action-list file has one column:

table_name
FscmTopModelAM.AccountBIAM.Account
FscmTopModelAM.InvoiceBIAM.Invoice

Developers can move table names between files. When a single manifest is needed, prepare it from the action folder:

manifest_file=dl.prepare_bicc_table_selection_manifest(action_folder=selection_dir)

To attach the action-list directory directly to the Data Load:

dl.prepare_data_load_tables(action_folder=selection_dir)created=dl.create_dataload()

Use BICC Load Options

Use BICCLoadOptions instead of remembering the raw dataLoadOptions JSON node or Knowledge Module option names.

from datatransforms.dataload_load_options import BICCLoadOptions

load_options = (
    BICCLoadOptions.incremental()
    .uppercase_names()
    .reserved_words_enclose_with_delimiters()
    .audit(job_id=True, operation=True, timestamp=False)
    .no_logging()
    .bicc_job_polling_interval(25)
    .bicc_job_timeout(1200)
    .cleanup_copy_data_logs(True)
    .conversion_errors(BICCLoadOptions.ConversionErrors.STORE_NULL)
    .show_copy_data_logs(BICCLoadOptions.ShowCopyDataLogs.ALWAYS)
    .reject_limit(0)
)

dl = DataLoad("demo_bicc_with_options", project_name="BICC_DEMO_PROJECT")

(
    dl.bicc_source(
        connection_name="DEMO_BICC",
        offerings=["Financial", "Service"])
    .incremental_merge(
        "FscmTopModelAM.AccountBIAM.Account",
        incremental_column=None,
        merge_keys=None)
    .target("tgt.TGT_USER")
    .load_options(load_options)
)

created = dl.create_dataload()

When using bicc_source and then load_options, the SDK preserves the selected offerings in dataLoadOptions unless the supplied options already contain offerings.

To produce only the JSON node:

data_load_options_json=load_options.to_json_node()

Complete Data Load example

Export offerings to CSV:

"""Example to export BICC offerings to a CSV file."""

from datatransforms.workbench import DataTransformsWorkbench, WorkbenchConfig


pswd = "<your deployment pswd from secret store>"
connect_params = WorkbenchConfig.get_workbench_config(pswd)

workbench = DataTransformsWorkbench()
workbench.connect_workbench(connect_params)

bicc_client = workbench.get_bicc_client()

output_file = bicc_client.export_matching_offerings_as_csv(
    connection_name="DEMO_BICC",
    output_file="bicc_offerings.csv")

print("BICC offerings exported to " + output_file)

Create a Data Load from exported offerings and participating tables:

"""Example to create a BICC Data Load from offerings and participating tables."""

from datatransforms.dataload import DataLoad
from datatransforms.dataload_load_options import BICCLoadOptions
from datatransforms.workbench import DataTransformsWorkbench, WorkbenchConfig


pswd = "<your deployment pswd from secret store>"
connect_params = WorkbenchConfig.get_workbench_config(pswd)

workbench = DataTransformsWorkbench()
workbench.connect_workbench(connect_params)

load_options = (
    BICCLoadOptions.incremental()
    .uppercase_names()
    .reserved_words_enclose_with_delimiters()
    .audit(job_id=True, operation=True, timestamp=False)
    .no_logging()
    .bicc_job_polling_interval(25)
    .bicc_job_timeout(1200)
    .cleanup_copy_data_logs(True)
    .conversion_errors(BICCLoadOptions.ConversionErrors.STORE_NULL)
    .show_copy_data_logs(BICCLoadOptions.ShowCopyDataLogs.ALWAYS)
    .reject_limit(0)
)

dl = DataLoad("dataLoadBICCFromOfferings", project_name="test")

(
    dl.bicc_source_from_offerings_csv(
        connection_name="DEMO_BICC",
        offerings_csv="bicc_offerings.csv")
    .prepare_data_load_tables(csv_file="bicc_table_selection.csv")
    .target("src.SRC_USER")
    .load_options(load_options)
)

created_dataload = dl.create_dataload()
print("BICC Data Load created: " + created_dataload["bulkLoadName"])