From 01bf4a65181750ff1c59f8e4f9e4561232e22990 Mon Sep 17 00:00:00 2001 From: Ardis Lu Date: Thu, 23 Jan 2025 19:47:14 -0800 Subject: [PATCH] Allow Array 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. --- examples/ex1_2.js | 6 +++--- examples/ex1_3.js | 4 ++-- examples/ex1_4.js | 2 +- examples/ex3_1.js | 2 +- examples/ex3_2.js | 2 +- src/core.js | 27 +++++++++++++++++++++------ test/core.test.js | 35 ++++++++++++++++++++++++++++++++++- 7 files changed, 63 insertions(+), 15 deletions(-) diff --git a/examples/ex1_2.js b/examples/ex1_2.js index c62a718..cf8a892 100644 --- 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 0e9e2fb..7fdc65b 100644 --- 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 2f32437..fa53b96 100644 --- 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 0f1610e..a8a0fbc 100644 --- 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 57e0fd4..caeadca 100644 --- 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 84486c6..1d70878 100644 --- 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|TransformationFunction} [handleHeaders=false] A `boolean`, `Array`, 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`, the header row will be replaced with the `Array`. 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 196dedf..52aba74 100644 --- 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', { 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> is returned', { concurrency: true }, async () => { await createCSVMockStream([ ['columnA', 'columnB'], -- 2.47.3