Http & Salesforce improvements

packages/http/CHANGELOG.md

@@ -1,5 +1,11 @@
 # @openfn/language-http
 
+## 6.1.0
+
+### Minor Changes
+
+- cfe1ccb: Add callback support for parseXML
+
 ## 6.0.0
 
 ### Major Changes

packages/http/ast.json

@@ -392,7 +392,8 @@
       "name": "parseXML",
       "params": [
         "body",
-        "script"
+        "script",
+        "callback"
       ],
       "docs": {
         "description": "Parse XML with the Cheerio parser",
@@ -404,7 +405,12 @@
           },
           {
             "title": "example",
-            "description": "parseXML(body, function($){\n   return $(\"table[class=your_table]\").parsetable(true, true, true);\n })"
+            "description": "parseXML(\n  (state) => state.response,\n  ($) => {\n    return $(\"table[class=your_table]\").parsetable(true, true, true);\n  }\n);"
+          },
+          {
+            "title": "example",
+            "description": "parseXML(\n  (state) => state.response,\n  ($) => {\n    return $(\"table[class=your_table]\").parsetable(true, true, true);\n  },\n  (next) => ({ ...next, results: next.data.body })\n);",
+            "caption": "Using parseXML with a callback"
           },
           {
             "title": "function",
@@ -429,6 +435,15 @@
             },
             "name": "script"
           },
+          {
+            "title": "param",
+            "description": "(Optional) Callback function",
+            "type": {
+              "type": "NameExpression",
+              "name": "function"
+            },
+            "name": "callback"
+          },
           {
             "title": "returns",
             "description": null,

packages/http/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@openfn/language-http",
-  "version": "6.0.0",
+  "version": "6.1.0",
   "description": "An HTTP request language package for use with Open Function",
   "homepage": "https://docs.openfn.org",
   "repository": {

packages/http/src/Adaptor.js

@@ -1,12 +1,5 @@
-import {
-  execute as commonExecute,
-  composeNextState,
-  expandReferences,
-} from '@openfn/language-common';
-import cheerio from 'cheerio';
-import cheerioTableparser from 'cheerio-tableparser';
-
-import { request as sendRequest } from './Utils';
+import { execute as commonExecute } from '@openfn/language-common';
+import { request as sendRequest, xmlParser } from './Utils';
 
 /**
  * Options provided to the HTTP request
@@ -161,32 +154,28 @@ export function del(path, params, callback) {
  * Parse XML with the Cheerio parser
  * @public
  * @example
- *  parseXML(body, function($){
- *    return $("table[class=your_table]").parsetable(true, true, true);
- *  })
+ *  parseXML(
+ *   (state) => state.response,
+ *   ($) => {
+ *     return $("table[class=your_table]").parsetable(true, true, true);
+ *   }
+ * );
+ * @example <caption>Using parseXML with a callback</caption>
+ *  parseXML(
+ *   (state) => state.response,
+ *   ($) => {
+ *     return $("table[class=your_table]").parsetable(true, true, true);
+ *   },
+ *   (next) => ({ ...next, results: next.data.body })
+ * );
  * @function
  * @param {String} body - data string to be parsed
  * @param {function} script - script for extracting data
+ * @param {function} callback - (Optional) Callback function
  * @returns {Operation}
  */
-export function parseXML(body, script) {
-  return state => {
-    const resolvedBody = expandReferences(body)(state);
-    const $ = cheerio.load(resolvedBody);
-    cheerioTableparser($);
-
-    if (script) {
-      const result = script($);
-      try {
-        const r = JSON.parse(result);
-        return composeNextState(state, r);
-      } catch (e) {
-        return composeNextState(state, { body: result });
-      }
-    } else {
-      return composeNextState(state, { body: resolvedBody });
-    }
-  };
+export function parseXML(body, script, callback) {
+  return xmlParser(body, script, callback);
 }
 
 export {

packages/http/src/Utils.js

@@ -6,6 +6,9 @@ import {
   makeBasicAuthHeader,
 } from '@openfn/language-common/util';
 
+import * as cheerio from 'cheerio';
+import cheerioTableparser from 'cheerio-tableparser';
+
 export function addBasicAuth(configuration = {}, headers) {
   const { username, password } = configuration;
   if (username && password) {
@@ -84,3 +87,23 @@ export function request(method, path, params, callback = s => s) {
       });
   };
 }
+
+export function xmlParser(body, script, callback = s => s) {
+  return state => {
+    const [resolvedBody] = expandReferences(state, body);
+    const $ = cheerio.load(resolvedBody);
+    cheerioTableparser($);
+
+    if (script) {
+      const result = script($);
+      try {
+        const r = JSON.parse(result);
+        return callback(composeNextState(state, r));
+      } catch (e) {
+        return callback(composeNextState(state, { body: result }));
+      }
+    } else {
+      return callback(composeNextState(state, { body: resolvedBody }));
+    }
+  };
+}

packages/salesforce/CHANGELOG.md

@@ -1,5 +1,11 @@
 # @openfn/language-salesforce
 
+## 4.6.0
+
+### Minor Changes
+
+- cfe1ccb: Add options and callback params in query function
+
 ## 4.5.2
 
 ### Patch Changes

packages/salesforce/ast.json

@@ -204,7 +204,9 @@
     {
       "name": "query",
       "params": [
-        "qs"
+        "qs",
+        "options",
+        "callback"
       ],
       "docs": {
         "description": "Execute an SOQL query.\nNote that in an event of a query error,\nerror logs will be printed but the operation will not throw the error.",
@@ -216,7 +218,12 @@
           },
           {
             "title": "example",
-            "description": "query(`SELECT Id FROM Patient__c WHERE Health_ID__c = '${state.data.field1}'`);"
+            "description": "query(state=> `SELECT Id FROM Patient__c WHERE Health_ID__c = '${state.data.field1}'`);"
+          },
+          {
+            "title": "example",
+            "description": "query(state=> `SELECT Id FROM Patient__c WHERE Health_ID__c = '${state.data.field1}'`, { autoFetch: true });",
+            "caption": "Query more records if next records are available"
           },
           {
             "title": "function",
@@ -232,6 +239,36 @@
             },
             "name": "qs"
           },
+          {
+            "title": "param",
+            "description": "Options passed to the bulk api.",
+            "type": {
+              "type": "NameExpression",
+              "name": "Object"
+            },
+            "name": "options"
+          },
+          {
+            "title": "param",
+            "description": "Fetch next records if available.",
+            "type": {
+              "type": "OptionalType",
+              "expression": {
+                "type": "NameExpression",
+                "name": "boolean"
+              }
+            },
+            "name": "options.autoFetch"
+          },
+          {
+            "title": "param",
+            "description": "A callback to execute once the record is retrieved",
+            "type": {
+              "type": "NameExpression",
+              "name": "Function"
+            },
+            "name": "callback"
+          },
           {
             "title": "returns",
             "description": null,
@@ -242,7 +279,7 @@
           }
         ]
       },
-      "valid": true
+      "valid": false
     },
     {
       "name": "bulkQuery",

packages/salesforce/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@openfn/language-salesforce",
-  "version": "4.5.2",
+  "version": "4.6.0",
   "description": "Salesforce Language Pack for OpenFn",
   "homepage": "https://docs.openfn.org",
   "exports": {

packages/salesforce/src/Adaptor.js

@@ -150,33 +150,82 @@ export function retrieve(sObject, id, callback) {
  * error logs will be printed but the operation will not throw the error.
  * @public
  * @example
- * query(`SELECT Id FROM Patient__c WHERE Health_ID__c = '${state.data.field1}'`);
+ * query(state=> `SELECT Id FROM Patient__c WHERE Health_ID__c = '${state.data.field1}'`);
+ * @example <caption>Query more records if next records are available</caption>
+ * query(state=> `SELECT Id FROM Patient__c WHERE Health_ID__c = '${state.data.field1}'`, { autoFetch: true });
  * @function
  * @param {String} qs - A query string.
+ * @param {Object} options - Options passed to the bulk api.
+ * @param {boolean} [options.autoFetch] - Fetch next records if available.
+ * @param {Function} callback - A callback to execute once the record is retrieved
  * @returns {Operation}
  */
-export function query(qs) {
-  return state => {
+export function query(qs, options, callback = s => s) {
+  return async state => {
+    let done = false;
+    let qResult = null;
+    let result = [];
+
     const { connection } = state;
-    const resolvedQs = expandReferences(qs)(state);
+    const [resolvedQs, resolvedOptions] = newExpandReferences(
+      state,
+      qs,
+      options
+    );
+    const { autoFetch } = { ...{ autoFetch: false }, ...resolvedOptions };
+
     console.log(`Executing query: ${resolvedQs}`);
+    try {
+      qResult = await connection.query(resolvedQs);
+    } catch (err) {
+      const { message, errorCode } = err;
+      console.log(`Error ${errorCode}: ${message}`);
+      throw err;
+    }
 
-    return connection.query(resolvedQs, function (err, result) {
-      if (err) {
-        const { message, errorCode } = err;
-        console.log(`Error ${errorCode}: ${message}`);
-        throw err;
+    if (qResult.totalSize > 0) {
+      console.log('Total records', qResult.totalSize);
+
+      while (!done) {
+        result.push(qResult);
+
+        if (qResult.done) {
+          done = true;
+        } else if (autoFetch) {
+          console.log(
+            'Fetched records so far',
+            result.map(ref => ref.records).flat().length
+          );
+          console.log('Fetching next records...');
+          try {
+            qResult = await connection.request({ url: qResult.nextRecordsUrl });
+          } catch (err) {
+            const { message, errorCode } = err;
+            console.log(`Error ${errorCode}: ${message}`);
+            throw err;
+          }
+        } else {
+          done = true;
+        }
       }
 
       console.log(
-        'Results retrieved and pushed to position [0] of the references array.'
+        'Done ✔ retrieved records',
+        result.map(ref => ref.records).flat().length
       );
+    } else {
+      console.log('No records found.');
+    }
 
-      return {
-        ...state,
-        references: [result, ...state.references],
-      };
-    });
+    console.log(
+      'Results retrieved and pushed to position [0] of the references array.'
+    );
+
+    const nextState = {
+      ...state,
+      references: [result, ...state.references],
+    };
+    return callback(nextState);
   };
 }