Integrator & ETL developer

Publish a SQL table as a JSON web service

A partner needs your product list. One script declares a typed GET service, queries the database through a stored connection and returns JSON. MentDB handles the URL, the authentication and the parameter checks.

script create get "api.products" false 1
  (param
    (var "[type]" {type is_char [type] 1} "Product type (A, B ...)" is_null:false is_empty:false "A")
  )
  "Return the products of one type as JSON"
{

	#1. Open the database connection;
	sql connect "session1" {cm get "demo_cm_mysql";};

	try {

		#2. Query and convert to JSON in one call;
		-> "[json]" (sql to json "session1" "products" (concat "select * from products where type=" (sql encode [type])));

	} {

		sql disconnect "session1";
		exception (1) [err];

	} "[err]";

	#3. Always release the connection;
	sql disconnect "session1";
	[json];

} "JSON with the products";

Import a CSV from SFTP into a database, every morning

A partner drops orders.csv on its SFTP server. The script downloads it, inserts each line in one transaction, writes a trace in the log, and a job runs it every day at 6:00.

script create exe "etl.import.orders" false 1
  (param
  ;)
  "Import the daily orders file from the partner SFTP"
{

	#1. Download the file;
	-> "[local]" (concat "tmp/orders_" (date sysdate) ".csv");
	sftp connect "sftp1" {cm get "demo_cm_sftp";};
	sftp get "sftp1" "/out/orders.csv" [local];
	sftp disconnect "sftp1";

	#2. Insert each line in one transaction;
	sql connect "db1" {cm get "demo_cm_mysql";};
	-> "[nb]" 0;
	csv parse (mql "O") [local] (mql ",") (mql "'") (mql "id,name,quantity") {
		sql dml "db1" (concat "insert into orders (id, name, quantity) values (" (sql encode [O_id]) ", " (sql encode [O_name]) ", " (sql encode [O_quantity]) ")");
		++ "[nb]";
	};
	sql commit "db1";
	sql disconnect "db1";

	#3. Trace the run;
	log write (concat [nb] " orders imported") OK "file" [local];

} "";

Mail a PDF report built from live data

The SQL result becomes an HTML table, the HTML becomes a PDF, and the PDF leaves as an attachment. The mail goes through the MentDB spooler, so a failed sending is kept and can be replayed.

script create exe "report.products.daily" false 1
  (param
  ;)
  "Send the daily products report"
{

	#1. Build the HTML table from SQL, then the PDF;
	-> "[pdf]" (concat "tmp/products_" (date sysdate) ".pdf");
	sql connect "db1" {cm get "demo_cm_mysql";};
	pdf from html (sql to html "db1" "products" (concat "select * from products")) [pdf];
	sql disconnect "db1";

	#2. Prepare the attachment list;
	json load "att" "[]";
	json iarray "att" / [pdf] STR;

	#3. Queue the mail (sent by the spooler, kept on error);
	mail send "demo_cm_smtp" "sales@example.com" "" "" "Daily products report" "Please find the report attached." (json doc "att");
	json unload "att";

} "";

Turn a partner CSV text into JSON objects

A partner sends a semicolon-separated text with a header line and a missing quantity. The script splits the lines, extracts each field with a regex, replaces the empty quantity by 0, computes the line amount and returns a JSON array ready for an API.

#1. CSV text received from a partner, separator semicolon;
-> "[csv]" (concat "sku;label;qty;price" [_n_] "BLT-M8;Bolt M8;250;0.12" [_n_] "NUT-M8;Nut M8;300;0.05" [_n_] "WSH-M8;Washer M8;;0.02");
#2. One JSON object per data line, empty quantity set to 0;
json load "rx_lines" (string split [csv] "\n" -1);
json load "rx_out" "[]";
json parse_array "rx_lines" "/" "[line]" {
	if (not (string starts_with [line] "sku;")) {
		-> "[qty]" (string regexp_extract [line] "^[^;]*;[^;]*;([^;]*);" 1);
		if (string is_blank [qty]) {-> "[qty]" 0;};
		json iarray "rx_out" "/" (concat "{\"sku\": \"" (string regexp_extract [line] "^([^;]*)" 1) "\", \"label\": \"" (string json_escape (string regexp_extract [line] "^[^;]*;([^;]*)" 1)) "\", \"qty\": " [qty] ", \"amount\": " (math round (* [qty] (string regexp_extract [line] "([^;]*)$" 1)) 2) "}") OBJ;
	};
};
#3. The JSON document ready for the API;
json doc "rx_out";

[
  {
    "amount": 30,
    "qty": 250,
    "label": "Bolt M8",
    "sku": "BLT-M8"
  },
  {
    "amount": 15,
    "qty": 300,
    "label": "Nut M8",
    "sku": "NUT-M8"
  },
  {
    "amount": 0,
    "qty": 0,
    "label": "Washer M8",
    "sku": "WSH-M8"
  }
]

Sort JSON orders and export them as CSV

An API returns orders as JSON; accounting wants a CSV sorted by amount with a total line. The script sorts the array, loops on the orders, quotes the values that contain the separator and adds the total.

#1. Orders as JSON from an API;
json load "rx_ord" "[{\"id\": \"A-102\", \"customer\": \"Dupont, SARL\", \"amount\": 1250.5}, {\"id\": \"A-103\", \"customer\": \"Martin\", \"amount\": 89.9}, {\"id\": \"A-104\", \"customer\": \"L'Atelier du Bois\", \"amount\": 640}]";
#2. Sort by amount, biggest first;
json sort "rx_ord" "/" "amount" "DESC";
#3. Write a CSV with a header and safely quoted values;
-> "[csv]" "id,customer,amount";
-> "[total]" 0;
json parse_array "rx_ord" "/" "[o]" {
	json load "rx_o" [o];
	-> "[csv]" (concat [csv] [_n_] (json select "rx_o" "/id") "," (string csv_value (json select "rx_o" "/customer") "," "\"") "," (json select "rx_o" "/amount"));
	-> "[total]" (+ [total] (json select "rx_o" "/amount"));
};
concat [csv] [_n_] "TOTAL,," [total];

id,customer,amount
A-102,"Dupont, SARL",1250.5
A-104,L'Atelier du Bois,640
A-103,Martin,89.9
TOTAL,,1980.4

Incremental sync between two databases with a watermark

Copy only the rows changed since the last run from an HSQL source to an H2 target. The last updated_at seen is kept in a parameter (the watermark); the target is upserted with MERGE, so a second run only moves the new and modified rows.

#1. Source and target;
cm set "tuto_p4i_src" {execute "db.hsql.embedded.config.get" "[database]" "tmp/rx_p4i_sync/src" "[user]" "SA" "[password]" "";};
cm set "tuto_p4i_dst" {execute "db.h2.embedded.config.get" "[database]" "./tmp/rx_p4i_sync/dst" "[user]" "sa" "[password]" "sa";};
sql connect "rx_p4i_s" {cm get "tuto_p4i_src"};
sql connect "rx_p4i_d" {cm get "tuto_p4i_dst"};
sql dml "rx_p4i_s" (concat "create table customer (id int primary key, name varchar(40), city varchar(30), updated_at timestamp)");
sql dml "rx_p4i_s" (concat "insert into customer values (1, 'Garage Hoarau', 'Saint-Denis', '2026-10-02 09:00:00'), (2, 'Pharmacie du Port', 'Le Port', '2026-10-05 14:30:00'), (3, 'Boulangerie Payet', 'Saint-Pierre', '2026-10-07 08:15:00')");
sql dml "rx_p4i_d" (concat "create table customer (id int primary key, name varchar(40), city varchar(30), updated_at timestamp)");
#2. The watermark, empty at the first run;
parameter add "tuto_p4i_wm" "1970-01-01 00:00:00" "Last updated_at copied to the target" false;
-> "[log]" "";
for (-> "[run]" 1) (<= [run] 2) (++ "[run]") {
	-> "[wm]" (parameter get value "tuto_p4i_wm");
	-> "[n]" 0;
	#3. Delta since the watermark, 4. upsert;
	sql parse "rx_p4i_s" "C" (concat "select id, name, city, updated_at from customer where updated_at > " (sql encode [wm]) " order by updated_at") {
		sql dml "rx_p4i_d" (concat "merge into customer key(id) values (" [C_ID] ", " (sql encode [C_NAME]) ", " (sql encode [C_CITY]) ", " (sql encode [C_UPDATED_AT]) ")");
		-> "[last]" [C_UPDATED_AT];
		++ "[n]";
	};
	#5. Advance the watermark;
	if (> [n] 0) {parameter update "tuto_p4i_wm" (string substring [last] 0 19);};
	-> "[log]" (concat [log] "run " [run] ": watermark " [wm] ", rows moved " [n] [_n_]);
	#Between the runs, the source changes: one update, one new customer;
	if (== [run] 1) {
		sql dml "rx_p4i_s" (concat "update customer set city = 'Saint-Paul', updated_at = '2026-10-20 10:00:00' where id = 2");
		sql dml "rx_p4i_s" (concat "insert into customer values (4, 'Fleurs du Lagon', 'Saint-Gilles', '2026-10-21 16:45:00')");
	};
};
-> "[target]" (sql to csv "rx_p4i_d" "customer" (concat "select id, name, city from customer order by id") ";" "\"");
-> "[final_wm]" (parameter get value "tuto_p4i_wm");
#Cleanup of the demo;
sql dml "rx_p4i_s" "SHUTDOWN";
sql disconnect "rx_p4i_s";
sql disconnect "rx_p4i_d";
cm remove "tuto_p4i_src";
cm remove "tuto_p4i_dst";
parameter remove "tuto_p4i_wm";
file delete "tmp/rx_p4i_sync";
concat [log] "watermark now: " [final_wm] [_n_] [target];

run 1: watermark 1970-01-01 00:00:00, rows moved 3
run 2: watermark 2026-10-07 08:15:00, rows moved 2
watermark now: 2026-10-21 16:45:00
id;name;city
1;Garage Hoarau;Saint-Denis
2;Pharmacie du Port;Saint-Paul
3;Boulangerie Payet;Saint-Pierre
4;Fleurs du Lagon;Saint-Gilles

One connection per environment and engine, checked without connecting

The integration platform targets several database servers: H2 and HSQL in server mode for the test benches, Derby (server and embedded) for an old Java application, MySQL in production. Declare them all, print the JDBC URL generated for each one, check they are all of type db, then remove them.

#1. The password of the test benches comes from a private parameter;
parameter add "tuto_p4i_bench_pwd" "Bench-2026!" "Password of the test bench databases" true;
-> "[pwd]" (parameter get value "tuto_p4i_bench_pwd");
cm set "tuto_p4i_h2srv" {execute "db.h2.server.config.get" "[hostname]" "bench1.lan.example.com" "[port]" "9092" "[database]" "orders" "[user]" "sa" "[password]" [pwd];};
cm set "tuto_p4i_hsqlsrv" {execute "db.hsql.server.config.get" "[hostname]" "bench2.lan.example.com" "[port]" "9001" "[database]" "stock" "[user]" "SA" "[password]" [pwd];};
cm set "tuto_p4i_derbysrv" {execute "db.derby.server.config.get" "[hostname]" "legacy.lan.example.com" "[port]" "1527" "[database]" "payroll" "[user]" "app" "[password]" [pwd];};
#2. The old Java application keeps its Derby database in a local folder;
cm set "tuto_p4i_derbyemb" {execute "db.derby.embedded.config.get" "[database]" "data/legacy/payroll" "[user]" "app" "[password]" [pwd];};
cm set "tuto_p4i_mysqlprd" {execute "db.mysql.config.get" "[hostname]" "mysql-prod.example.com" "[port]" "3306" "[database]" "erp" "[user]" "integration" "[password]" [pwd];};
-> "[pwd]" "";
#3. Check each one;
-> "[rep]" "";
json load "rx_p4i_ids" "[\"tuto_p4i_h2srv\",\"tuto_p4i_hsqlsrv\",\"tuto_p4i_derbysrv\",\"tuto_p4i_derbyemb\",\"tuto_p4i_mysqlprd\"]";
json parse_array "rx_p4i_ids" "/" "[id]" {
	json load "rx_p4i_cm" (cm get [id]);
	-> "[rep]" (concat [rep] (string rpad (string replace [id] "tuto_p4i_" "") " " 10) " db=" (cm exist [id] "db") "  " (json select "rx_p4i_cm" "/jdbc") [_n_]);
};
#4. Inventory and cleanup;
-> "[count]" (string regexp_count (cm show "db") "tuto_p4i_");
json parse_array "rx_p4i_ids" "/" "[id]" {cm remove [id];};
parameter remove "tuto_p4i_bench_pwd";
json unload "rx_p4i_ids";
json unload "rx_p4i_cm";
concat [rep] "Declared db connections: " [count] ", left after cleanup: " (string regexp_count (cm show "db") "tuto_p4i_");

h2srv      db=1  jdbc:h2:tcp://bench1.lan.example.com:9092/orders
hsqlsrv    db=1  jdbc:hsqldb:hsql:bench2.lan.example.com:9001/stock
derbysrv   db=1  jdbc:derby://legacy.lan.example.com:1527/payroll
derbyemb   db=1  jdbc:derby:data/legacy/payroll
mysqlprd   db=1  jdbc:mysql://mysql-prod.example.com:3306/erp
Declared db connections: 5, left after cleanup: 0

Apply a partner's CSV delta to a price table with MERGE

Each night a supplier sends a CSV with new and changed prices. Apply it to the local price table with an upsert, count the inserts and the updates, reject the lines with an invalid price and give the variation of each changed price.

#1. The local table;
cm set "tuto_p4i_price_db" {execute "db.h2.embedded.config.get" "[database]" "./tmp/rx_p4i_prices/db" "[user]" "sa" "[password]" "sa";};
sql connect "rx_p4i_p" {cm get "tuto_p4i_price_db"};
sql dml "rx_p4i_p" (concat "create table price (sku varchar(10) primary key, label varchar(40), price decimal(10,2), old_price decimal(10,2))");
sql dml "rx_p4i_p" (concat "insert into price values ('A-100', 'Pallet wrap', 18.50, null), ('A-205', 'Carton 40x30', 0.62, null), ('B-310', 'Label roll', 7.90, null)");
#2. The delta of the night;
file create "tmp/rx_p4i_delta.csv" (concat "sku;label;price" [_n_] "A-205;Carton 40x30;0.66" [_n_] "B-310;Label roll;7.50" [_n_] "C-020;Tape 50mm;1.35" [_n_] "C-044;Strapping band;N/A") "UTF-8";
-> "[ins]" 0;
-> "[upd]" 0;
-> "[rej]" "";
csv parse (mql "D") (mql "tmp/rx_p4i_delta.csv") (mql ";") (mql "\"") (mql "") {
	if (type is_number [D_price]) {
		-> "[exists]" (sql value "rx_p4i_p" (concat "select count(*) from price where sku = " (sql encode [D_sku])));
		if (== [exists] 0) {++ "[ins]"; -> "[old]" "null";} {++ "[upd]"; -> "[old]" (sql value "rx_p4i_p" (concat "select price from price where sku = " (sql encode [D_sku])));};
		#3. Upsert, the old price is kept for the report;
		sql dml "rx_p4i_p" (concat "merge into price key(sku) values (" (sql encode [D_sku]) ", " (sql encode [D_label]) ", " [D_price] ", " [old] ")");
	} {
		-> "[rej]" (concat [rej] [D_sku] " (price " [D_price] ") ");
	};
};
#4. Report;
json load "rx_p4i_var" (sql to json "rx_p4i_p" "variation" (concat "select sku, old_price, price, round(100*(price-old_price)/old_price, 1) as pct from price where old_price is not null order by sku"));
-> "[res]" (concat "inserted: " [ins] ", updated: " [upd] ", rejected: " [rej] [_n_]);
json parse_array "rx_p4i_var" "/data" "[v]" {
	json load "rx_p4i_v" [v];
	-> "[res]" (concat [res] (json select "rx_p4i_v" "/sku") " " (json select "rx_p4i_v" "/old_price") " -> " (json select "rx_p4i_v" "/price") " (" (json select "rx_p4i_v" "/pct") " %)" [_n_]);
};
-> "[res]" (concat [res] "rows in the table: " (sql value "rx_p4i_p" (concat "select count(*) from price")));
sql disconnect "rx_p4i_p";
cm remove "tuto_p4i_price_db";
file delete "tmp/rx_p4i_prices";
file delete "tmp/rx_p4i_delta.csv";
json unload "rx_p4i_var";
json unload "rx_p4i_v";
[res];

inserted: 1, updated: 2, rejected: C-044 (price N/A)
A-205 0.62 -> 0.66 (6.5 %)
B-310 7.90 -> 7.50 (-5.1 %)
rows in the table: 4

Nightly export pack: CSV, Excel and PDF in an outbox connection

Produce the nightly export pack for the accounting team from the orders database: a CSV for their tool, an Excel file for the controllers and a PDF listing, written into the folder of a file connection, then zipped for the transfer.

#1. The database and the outbox;
cm set "tuto_p4i_ord_db" {execute "db.h2.embedded.config.get" "[database]" "./tmp/rx_p4i_exp/db" "[user]" "sa" "[password]" "sa";};
cm set "tuto_p4i_outbox" {execute "file.local.config.get" "[dir]" "tmp/rx_p4i_outbox";};
json load "rx_p4i_ob" (cm get "tuto_p4i_outbox");
-> "[out]" (json select "rx_p4i_ob" "/localDir");
file mkdir [out];
sql connect "rx_p4i_e" {cm get "tuto_p4i_ord_db"};
sql dml "rx_p4i_e" (concat "create table orders (id int primary key, order_day date, customer varchar(40), amount decimal(10,2))");
sql dml "rx_p4i_e" (concat "insert into orders values (7001, '2026-10-31', 'Garage Hoarau', 420.00), (7002, '2026-10-31', 'Pharmacie du Port', 89.90), (7003, '2026-10-31', 'Garage Hoarau', 1310.50), (7004, '2026-10-31', 'Fleurs du Lagon', 56.00)");
-> "[q]" (concat "select id, order_day, customer, amount from orders where order_day = '2026-10-31' order by id");
#2. CSV, Excel and PDF from the same query;
sql to csv_file "rx_p4i_e" (concat [out] "/orders_20261031.csv") [q] ";" "\"";
sql to excelx "rx_p4i_e" "orders" [q] (concat [out] "/orders_20261031.xlsx");
sql to pdf "rx_p4i_e" "orders" [q] (concat [out] "/orders_20261031.pdf");
-> "[total]" (sql value "rx_p4i_e" (concat "select sum(amount) from orders where order_day = '2026-10-31'"));
sql disconnect "rx_p4i_e";
#3. Pack the outbox;
compress zip [out] "tmp/rx_p4i_pack_20261031.zip";
-> "[rep]" "";
json load "rx_p4i_fl" (file dir_list_regex [out] "orders_.*" true false);
json parse_array "rx_p4i_fl" "/" "[f]" {
	-> "[rep]" (concat [rep] [f] " " (file size (concat [out] "/" [f])) " B" [_n_]);
};
-> "[rep]" (concat [rep] "CSV lines: " (file count_lines (concat [out] "/orders_20261031.csv")) ", PDF pages: " (pdf page_count_file (concat [out] "/orders_20261031.pdf")) ", total " [total] " EUR" [_n_] "Pack: " (file size "tmp/rx_p4i_pack_20261031.zip") " B");
#4. Cleanup of the demo;
cm remove "tuto_p4i_ord_db";
cm remove "tuto_p4i_outbox";
file delete [out];
file delete "tmp/rx_p4i_exp";
file delete "tmp/rx_p4i_pack_20261031.zip";
json unload "rx_p4i_ob";
json unload "rx_p4i_fl";
[rep];

orders_20261031.xlsx 3508 B
orders_20261031.csv 181 B
orders_20261031.pdf 2414 B
CSV lines: 5, PDF pages: 1, total 1876.40 EUR
Pack: 5550 B

Nightly copy of an Oracle table into MySQL, committed by batches

Copy the day's invoice lines from the Oracle finance database into the MySQL reporting database: read with sql parse, insert in MySQL, commit every 500 rows to keep the transactions short, and log the volume and the duration.

#1. Connections (declared once on the server, here for the example);
cm set "tuto_p4i_fin" {execute "db.oracle.config.get" "[hostname]" "ora.erp.example.com" "[port]" "1521" "[database]" "FINPRD" "[user]" "fin_read" "[password]" (parameter get value "tuto_p4i_fin_pwd");};
cm set "tuto_p4i_rep" {execute "db.mysql.config.get" "[hostname]" "mysql-rep.example.com" "[port]" "3306" "[database]" "reporting" "[user]" "loader" "[password]" (parameter get value "tuto_p4i_rep_pwd");};
-> "[t0]" (date systimestamp);
sql connect "rx_p4i_ora" {cm get "tuto_p4i_fin";};
sql connect "rx_p4i_my" {cm get "tuto_p4i_rep";};
sql set_timeout 600;
#2. Short transactions on the target;
sql auto_commit "rx_p4i_my" false;
sql dml "rx_p4i_my" (concat "delete from invoice_line where inv_date = " (sql encode (date sysdate)));
#3. Copy, commit every 500 rows;
-> "[n]" 0;
sql parse "rx_p4i_ora" "L" (concat "select inv_no, line_no, account, amount, inv_date from fin.invoice_line where inv_date = trunc(sysdate)") {
	sql dml "rx_p4i_my" (concat "insert into invoice_line values (" (sql encode [L_INV_NO]) ", " [L_LINE_NO] ", " (sql encode [L_ACCOUNT]) ", " [L_AMOUNT] ", " (sql encode (string substring [L_INV_DATE] 0 10)) ")");
	++ "[n]";
	if (== (math mod [n] 500) 0) {sql commit "rx_p4i_my";};
};
sql commit "rx_p4i_my";
sql disconnect "rx_p4i_ora";
sql disconnect "rx_p4i_my";
#4. Trace;
log write "rx_p4i_nightly" (concat [n] " invoice lines copied in " (date duration [t0] (date systimestamp)));
[n];