Allow Array<any> and TransformationFunction as inputs to handle headers

This change simplifies the common tasks of (1) renaming headers (just pass the new header array) and (2) appending headers (pass a one-line TransformationFunction). Setting the option to a boolean is the same behavior as includeHeaders.

diff --git a/examples/ex1_2.js b/examples/ex1_2.js
index c62a718b9199dc51545da67224855d163fed4c7a..cf8a892a6ba442f35496c3bda313af6660dd3005 100644 (file)
--- a/examples/ex1_2.js
+++ b/examples/ex1_2.js
@@ -1,10 +1,10 @@
 // ## Example 1.2: Updating headers in the output
 // 
-// Set `includeHeaders: true` to manually handle the header row (first row).
+// Set `handleHeaders: true` to manually handle the header row (first row).
 // 
 // Notes:
 // - The header row is assumed to be the first row in the input CSV, and by default it goes directly to the output (skipping the user-provided transformation function).
-// - If your input CSV does not have a header row, set `includeHeaders: true` and process the header row (first row) normally.
+// - If your input CSV does not have a header row, set `handleHeaders: true` and process the header row (first row) normally.
 // - Otherwise, you will need distinct logic to handle the header row vs. data rows. A simple boolean variable is a good way to switch between this logic. 
 
 import { CSVReader, CSVTransformer, CSVWriter } from '../src/index.js';
@@ -19,5 +19,5 @@ function sum(row) {
 }
 
 await new CSVReader(new URL('./data/ex1_2-in.csv', import.meta.url))
-  .pipeThrough(new CSVTransformer(sum, { includeHeaders: true }))
+  .pipeThrough(new CSVTransformer(sum, { handleHeaders: true }))
   .pipeTo(new CSVWriter(new URL('./data/ex1_2-out.csv', import.meta.url)));
\ No newline at end of file

diff --git a/examples/ex1_3.js b/examples/ex1_3.js
index 0e9e2fbf14d9f2e5ff19f3ac000e1bacb333bd23..7fdc65b23c5dd035ef2669edc283b2e8cd769553 100644 (file)
--- a/examples/ex1_3.js
+++ b/examples/ex1_3.js
@@ -23,6 +23,6 @@ function calculate(obj) {
 }
 
 await new CSVReader(new URL('./data/ex1_3-in.csv', import.meta.url))
-  .pipeThrough(new CSVTransformer(parse, { includeHeaders: true, rawOutput: true }))
-  .pipeThrough(new CSVTransformer(calculate, { includeHeaders: true, rawInput: true }))
+  .pipeThrough(new CSVTransformer(parse, { handleHeaders: true, rawOutput: true }))
+  .pipeThrough(new CSVTransformer(calculate, { handleHeaders: true, rawInput: true }))
   .pipeTo(new CSVWriter(new URL('./data/ex1_3-out.csv', import.meta.url)));
\ No newline at end of file

diff --git a/examples/ex1_4.js b/examples/ex1_4.js
index 2f324379a0eeeb8c5b9074076645c03f8b2894cc..fa53b9695d77c961d45dbf72d48b18c0f43852eb 100644 (file)
--- a/examples/ex1_4.js
+++ b/examples/ex1_4.js
@@ -25,5 +25,5 @@ function parse(row) {
 }
 
 await new CSVReader(new URL('./data/ex1_4-in.csv', import.meta.url))
-  .pipeThrough(new CSVTransformer(parse, { includeHeaders: true }))
+  .pipeThrough(new CSVTransformer(parse, { handleHeaders: true }))
   .pipeTo(new CSVWriter(new URL('./data/ex1_4-out.csv', import.meta.url)));
\ No newline at end of file

diff --git a/examples/ex3_1.js b/examples/ex3_1.js
index 0f1610e31ff4638100f140ff8278ccf783df5238..a8a0fbcc8bd771ae79b0c5a3be76cd226e9adc45 100644 (file)
--- a/examples/ex3_1.js
+++ b/examples/ex3_1.js
@@ -29,5 +29,5 @@ function process(batch) {
 }
 
 await new CSVReader(new URL('./data/ex3_1-in.csv', import.meta.url))
-  .pipeThrough(new CSVTransformer(process, { includeHeaders: true, maxBatchSize: 5 }))
+  .pipeThrough(new CSVTransformer(process, { handleHeaders: true, maxBatchSize: 5 }))
   .pipeTo(new CSVWriter(new URL('./data/ex3_1-out.csv', import.meta.url)));
\ No newline at end of file

diff --git a/examples/ex3_2.js b/examples/ex3_2.js
index 57e0fd4e287759c4c1e38d3ce64e7856dd599c0e..caeadcaa9b216cf723efd145ab55677c2281fada 100644 (file)
--- a/examples/ex3_2.js
+++ b/examples/ex3_2.js
@@ -28,5 +28,5 @@ async function concurrent(row) {
 }
 
 await new CSVReader(new URL('./data/ex3_2-in.csv', import.meta.url))
-  .pipeThrough(new CSVTransformer(concurrent, { includeHeaders: true, maxConcurrent: 5 }))
+  .pipeThrough(new CSVTransformer(concurrent, { handleHeaders: true, maxConcurrent: 5 }))
   .pipeTo(new CSVWriter(new URL('./data/ex3_2-out.csv', import.meta.url)));
\ No newline at end of file

diff --git a/src/core.js b/src/core.js
index 84486c64ff997310d0c5d2e05fbcfc3cc037cbf2..1d708781d652b652ef6c3f5796af404121719761 100644 (file)
--- a/src/core.js
+++ b/src/core.js
@@ -157,10 +157,12 @@ export class CSVReader extends ReadableStream {
 /**
  * Options to configure `CSVTransformer`.
  * @typedef {Object} CSVTransformerOptions
- * @property {boolean} [includeHeaders=false] Set to `true` to pass the header row (assumed to be the first row of the CSV) to `fn()`. Otherwise,
- * the header row will flow through to the next stream without going through `fn()`. The default value is `false`.
+ * @property {boolean|Array<any>|TransformationFunction} [handleHeaders=false] A `boolean`, `Array<any>`, or `TransformationFunction` to process
+ * the header row (first row) of the CSV. If `false`, the header row will pass through the stream. If `true`, the header row will be passed to
+ * `fn()`. If `Array<any>`, the header row will be replaced with the `Array<any>`. If `TransformationFunction`, the header row will be passed
+ * to the `TransformationFunction`. The default value is `false`.
  * 
- * NOTE: If the CSV has no header row, set `includeHeaders` to `true` and process the first row normally in `fn()`.
+ * NOTE: If the CSV has no header row, set `handleHeaders` to `true` and process the first row normally in `fn()`.
  * @property {boolean} [rawInput=false] Set to `true` to send the raw CSV row to `fn()` as a `string`. Otherwise, the CSV row will be deserialized
  * using `JSON.parse()` before being sent to `fn()`. The default value is `false`.
  * @property {boolean} [rawOutput=false] Set to `true` to send the raw return value of `fn()` to the next stream. Otherwise, the return value of
@@ -214,7 +216,7 @@ export class CSVTransformer extends TransformStream {
       flush: (controller) => this.#flush(controller),
     });
 
-    options.includeHeaders ??= false;
+    options.handleHeaders ??= false;
     options.rawInput ??= false;
     options.rawOutput ??= false;
     options.onError ??= null;
@@ -225,6 +227,19 @@ export class CSVTransformer extends TransformStream {
     this.#options = options;
   }
 
+  async #wrappedHeaderFn(row) {
+    const { onError, handleHeaders } = this.#options;
+    if (typeof handleHeaders === 'boolean') {
+      return handleHeaders ? await this.#wrappedFn(row) : row;
+    }
+    if (Array.isArray(handleHeaders)) {
+      return handleHeaders;
+    }
+    if (onError === null) { return await handleHeaders(row); }
+    try { return await handleHeaders(row); }
+    catch (e) { return await onError(row, e, handleHeaders); }
+  }
+
   async #wrappedFn(row) {
     const { onError } = this.#options;
     if (onError === null) { return await this.#fn(row); }
@@ -267,12 +282,12 @@ export class CSVTransformer extends TransformStream {
   }
 
   async #transform(chunk, controller) {
-    const { includeHeaders, rawInput, maxBatchSize, maxConcurrent } = this.#options;
+    const { rawInput, maxBatchSize, maxConcurrent } = this.#options;
     const row = rawInput ? chunk : JSON.parse(chunk);
 
     if (this.#firstChunk) {
       this.#firstChunk = false;
-      const out = includeHeaders ? await this.#wrappedFn(row) : row;
+      const out = await this.#wrappedHeaderFn(row);
       this.#enqueueRow(out, controller);
       return;
     }

diff --git a/test/core.test.js b/test/core.test.js
index 196dedf9f27a5eef60e71bc220cb13eeb5a4df61..52aba74fe3b1a4232cc19e9fbf5caa0c2185f778 100644 (file)
--- a/test/core.test.js
+++ b/test/core.test.js
@@ -216,7 +216,7 @@ suite('CSVTransformer', { concurrency: true }, () => {
       ['columnA', 'columnB'],
       ['a', 'b']
     ])
-      .pipeThrough(new CSVTransformer(() => '["abc", "def"]', { includeHeaders: true, rawOutput: true }))
+      .pipeThrough(new CSVTransformer(() => '["abc", "def"]', { handleHeaders: true, rawOutput: true }))
       .pipeTo(csvStreamEqualWritable([
         ['abc', 'def'],
         ['abc', 'def']
@@ -241,6 +241,39 @@ suite('CSVTransformer', { concurrency: true }, () => {
         ['a', 'b'],
       ]));
   });
+  test('passes through header row when handleHeaders is true', { concurrency: true }, async () => {
+    await createCSVMockStream([
+      ['a', 'b'],
+      ['1', '2']
+    ])
+      .pipeThrough(new CSVTransformer(r => r.map(f => `${f}-modified`), { handleHeaders: true }))
+      .pipeTo(csvStreamEqualWritable([
+        ['a-modified', 'b-modified'],
+        ['1-modified', '2-modified']
+      ]));
+  });
+  test('replaces header row when handleHeaders is Array<any>', { concurrency: true }, async () => {
+    await createCSVMockStream([
+      ['a', 'b'],
+      ['1', '2']
+    ])
+      .pipeThrough(new CSVTransformer(r => r, { handleHeaders: ['aaa', 'bbb'] }))
+      .pipeTo(csvStreamEqualWritable([
+        ['aaa', 'bbb'],
+        ['1', '2']
+      ]));
+  });
+  test('processes header row when handleHeaders is TransformationFunction', { concurrency: true }, async () => {
+    await createCSVMockStream([
+      ['a', 'b'],
+      ['1', '2']
+    ])
+      .pipeThrough(new CSVTransformer(r => r, { handleHeaders: h => [...h, 'new header'] }))
+      .pipeTo(csvStreamEqualWritable([
+        ['a', 'b', 'new header'],
+        ['1', '2']
+      ]));
+  });
   test('enqueues multiple rows when Array<Array<any>> is returned', { concurrency: true }, async () => {
     await createCSVMockStream([
       ['columnA', 'columnB'],