From 1a67489b9e9e17119d91a82ee388c5550eeb4fb4 Mon Sep 17 00:00:00 2001 From: Ardis Lu Date: Wed, 9 Jul 2025 22:10:20 -0700 Subject: [PATCH] Ignore `handleHeaders` if the row has been normalized This behavior is desirable when `CSVTransformer` interacts with `CSVNormalizer` or any other transformation which converts the row from `Array` into `Array`, because it avoids the need to pass `handleHeaders: true` to `CSVTransformer` (which is counterintuitive). It is more accurate to say the header row has been deleted by the transformation and must by re-inserted into the write stream later. The loosening of the type checking is not a breaking change. Since the row is passed directly to `JSON.parse()`, the fields could have always been type `Object` instead of `string`. --- src/core.js | 15 +++++++++++---- test/core.test.js | 11 +++++++++++ test/normalization.test.js | 21 ++++++++++++++++++++- 3 files changed, 42 insertions(+), 5 deletions(-) diff --git a/src/core.js b/src/core.js index 5b6f024..9594899 100644 --- a/src/core.js +++ b/src/core.js @@ -202,7 +202,7 @@ export class CSVReader extends ReadableStream { /** * A function to process a row of CSV data from `CSVReader`. * @callback TransformationFunction - * @param {Array} row A row of input CSV data before transformation. + * @param {Array} row A row of input CSV data before transformation. * @returns {TransformationOutput|Promise} */ @@ -216,14 +216,14 @@ export class CSVReader extends ReadableStream { /** * A function to process multiple rows of CSV data from `CSVReader`. * @callback TransformationFunctionBatch - * @param {Array>} rows Multiple rows of input CSV data. + * @param {Array>} rows Multiple rows of input CSV data. * @returns {TransformationOutput|Promise} */ /** * A function to handle errors thrown by a transformation function. * @callback TransformationErrorFunction - * @param {Array} row The row passed to the transformation function which threw the error. + * @param {Array} row The row passed to the transformation function which threw the error. * @param {Error} error The error thrown by the transformation function. * @param {TransformationFunction} fn The transformation function itself. This argument can be used to retry a transformation. * @returns {TransformationOutput|Promise} @@ -241,7 +241,7 @@ export class CSVReader extends ReadableStream { /** * A function to handle errors thrown by a transformation function. * @callback TransformationErrorFunctionBatch - * @param {Array>} rows The rows of input CSV data passed to the transformation function which threw the error. + * @param {Array>} rows The rows of input CSV data passed to the transformation function which threw the error. * @param {Error} error The error thrown by the transformation function. * @param {TransformationFunctionBatch} fn The transformation function itself. This argument can be used to retry a transformation. * @returns {TransformationOutput|Promise} @@ -433,6 +433,13 @@ export class CSVTransformer extends TransformStream { async #transform(chunk, controller) { const row = this.#rawInput ? chunk : JSON.parse(chunk); + // If row is an array of objects, then the transformation has "normalized" the data and there is no header row. + // Instead, the header data has been incorporated into the object keys. The transformation is responsible + // for "denormalizing" the object back into an (extra) header row in the output. Ignore `handleHeaders`. + if (typeof row[0] === 'object') { + this.#firstChunk = false; + } + if (this.#firstChunk) { this.#firstChunk = false; const out = await this.#wrappedHeaderFn(row); diff --git a/test/core.test.js b/test/core.test.js index 3e0180e..e5100d0 100644 --- a/test/core.test.js +++ b/test/core.test.js @@ -400,6 +400,17 @@ suite('CSVTransformer', { concurrency: true }, () => { ['1', '2'] ])); }); + test('ignores handleHeaders when row is an array of objects (normalized)', { concurrency: true, only: true }, async () => { + await createCSVMockStream([ + [{ value: 'a1' }, { value: 'b1' }], + [{ value: 'a2' }, { value: 'b2' }], + ]) + .pipeThrough(new CSVTransformer(r => r.map(f => f.value), { handleHeaders: ['updatedA', 'updatedB'] })) + .pipeTo(csvStreamEqualWritable([ + ['a1', 'b1'], + ['a2', 'b2'] + ])); + }); test('enqueues multiple rows when Array> is returned', { concurrency: true }, async () => { await createCSVMockStream([ ['columnA', 'columnB'], diff --git a/test/normalization.test.js b/test/normalization.test.js index c12577c..3ee5ecf 100644 --- a/test/normalization.test.js +++ b/test/normalization.test.js @@ -2,7 +2,7 @@ import { suite, test } from 'node:test'; import { ok, deepStrictEqual, throws } from 'node:assert/strict'; import { csvStreamEqualWritable, createCSVMockStream, assertConsole } from './utils.js'; -import { CSVReader } from '../src/core.js'; +import { CSVReader, CSVTransformer } from '../src/core.js'; import { getDecimalSeparator, toCamelCase, expandScientificNotation, CSVNormalizer, CSVDenormalizer } from '../src/normalization.js'; suite('getDecimalSeparator', { concurrency: true }, () => { @@ -397,4 +397,23 @@ suite('CSVNormalizer and CSVDenormalizer end-to-end', { concurrency: true }, () ['N/A', 123100, '1000000000000000000', '2024-12-31T08:00:00.000Z'] ])); }); + test('can normalize with transformation', { concurrency: true }, async () => { + const headers = [ + /** @type {const} */({ name: 'columnA' }), + /** @type {const} */({ name: 'columnB' }) + ] + await createCSVMockStream([ + ['columnA', '', 'columnB', '', ''], + ['a', '', 'b', '', ''], + ['', '', '', '', ''], + ['', '', '', '', ''] + ]) + .pipeThrough(new CSVNormalizer(headers)) + .pipeThrough(new CSVTransformer(r => r.map(f => ({ name: `${f.name} updated`, value: `${f.value} updated` })))) + .pipeThrough(new CSVDenormalizer()) + .pipeTo(csvStreamEqualWritable([ + ['columnA updated', 'columnB updated'], + ['a updated', 'b updated'] + ])); + }); }); \ No newline at end of file -- 2.47.3