From: Ardis Lu Date: Sun, 19 Jan 2025 06:22:29 +0000 (-0800) Subject: Refactor createCSVTransformStream to CSVTransformer X-Git-Tag: v0.1.0~118 X-Git-Url: https://git.ardis.lu/?a=commitdiff_plain;h=5124850d4f02bc49a3305cc8cd9d22783e046db9;p=csvio Refactor createCSVTransformStream to CSVTransformer Using classes is more aligned with existing stream APIs such as TextEncoder and CompressionStream. --- diff --git a/README.md b/README.md index 021655d..17b2a20 100644 --- a/README.md +++ b/README.md @@ -20,7 +20,7 @@ Run: ```javascript import { createCSVReadableStream, - createCSVTransformStream, + CSVTransformer, createCSVWritableStream } from './src/index.js'; @@ -29,7 +29,7 @@ function timesTwo(row) { } await createCSVReadableStream('./data/example-in.csv') - .pipeThrough(createCSVTransformStream(timesTwo)) + .pipeThrough(new CSVTransformer(timesTwo)) .pipeTo(createCSVWritableStream('./data/example-out.csv')); ``` @@ -79,18 +79,18 @@ import { createCSVReadableStream } from './src/index.js'; const readableStream = createCSVReadableStream('./data/example-in.csv'); ``` -### `createCSVTransformStream` +### `CSVTransformer` -Create a `TransformStream` to apply a given function to each row in a streamed CSV. +A `TransformStream` that will apply a given function to each row in a streamed CSV. ```javascript -import { createCSVTransformStream } from './src/index.js'; +import { CSVTransformer } from './src/index.js'; function timesTwo(row) { return [Number(row[0]) * 2, Number(row[1]) * 2]; } -const transformStream = createCSVTransformStream(timesTwo); +const transformStream = new CSVTransformer(timesTwo); ``` ### `createCSVWritableStream` diff --git a/examples/ex1.js b/examples/ex1.js index d08d007..13f212c 100644 --- a/examples/ex1.js +++ b/examples/ex1.js @@ -8,12 +8,12 @@ // - Each call to the transformation function passes 1 row of the input CSV. // - The expected output of the transformation function is another array representing 1 row of the output CSV. -import { createCSVReadableStream, createCSVTransformStream, createCSVWritableStream } from '../src/index.js'; +import { createCSVReadableStream, CSVTransformer, createCSVWritableStream } from '../src/index.js'; function timesTwo(row) { return [Number(row[0]) * 2, Number(row[1]) * 2]; } await createCSVReadableStream(new URL('./data/ex1-in.csv', import.meta.url)) - .pipeThrough(createCSVTransformStream(timesTwo)) + .pipeThrough(new CSVTransformer(timesTwo)) .pipeTo(createCSVWritableStream(new URL('./data/ex1-out.csv', import.meta.url))); \ No newline at end of file diff --git a/examples/ex2.js b/examples/ex2.js index dfedb32..ac435b7 100644 --- a/examples/ex2.js +++ b/examples/ex2.js @@ -7,7 +7,7 @@ // - If your input CSV does not have a header row, set `includeHeaders: 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 { createCSVReadableStream, createCSVTransformStream, createCSVWritableStream } from '../src/index.js'; +import { createCSVReadableStream, CSVTransformer, createCSVWritableStream } from '../src/index.js'; let firstRow = true; function sum(row) { @@ -19,5 +19,5 @@ function sum(row) { } await createCSVReadableStream(new URL('./data/ex2-in.csv', import.meta.url)) - .pipeThrough(createCSVTransformStream(sum, { includeHeaders: true })) + .pipeThrough(new CSVTransformer(sum, { includeHeaders: true })) .pipeTo(createCSVWritableStream(new URL('./data/ex2-out.csv', import.meta.url))); \ No newline at end of file diff --git a/examples/ex3.js b/examples/ex3.js index 85a3c82..95c0f28 100644 --- a/examples/ex3.js +++ b/examples/ex3.js @@ -5,7 +5,7 @@ // transformations (e.g., transformations spread across multiple functions or lower-level transformations directly // on byte arrays). -import { createCSVReadableStream, createCSVTransformStream, createCSVWritableStream } from '../src/index.js'; +import { createCSVReadableStream, CSVTransformer, createCSVWritableStream } from '../src/index.js'; // Parse the raw CSV data and transform the data into an object for convenience function parse(row) { @@ -23,6 +23,6 @@ function calculate(obj) { } await createCSVReadableStream(new URL('./data/ex3-in.csv', import.meta.url)) - .pipeThrough(createCSVTransformStream(parse, { includeHeaders: true, rawOutput: true })) - .pipeThrough(createCSVTransformStream(calculate, { includeHeaders: true, rawInput: true })) + .pipeThrough(new CSVTransformer(parse, { includeHeaders: true, rawOutput: true })) + .pipeThrough(new CSVTransformer(calculate, { includeHeaders: true, rawInput: true })) .pipeTo(createCSVWritableStream(new URL('./data/ex3-out.csv', import.meta.url))); \ No newline at end of file diff --git a/examples/ex5.js b/examples/ex5.js index 7d8b5eb..ab7c762 100644 --- a/examples/ex5.js +++ b/examples/ex5.js @@ -5,7 +5,7 @@ // // Return `null` in the transformation function to consume an input row without emitting an output row. -import { createCSVReadableStream, createCSVTransformStream, createCSVWritableStream } from '../src/index.js'; +import { createCSVReadableStream, CSVTransformer, createCSVWritableStream } from '../src/index.js'; let i = 0; let report, id, date; @@ -25,5 +25,5 @@ function parse(row) { } await createCSVReadableStream(new URL('./data/ex5-in.csv', import.meta.url)) - .pipeThrough(createCSVTransformStream(parse, { includeHeaders: true })) + .pipeThrough(new CSVTransformer(parse, { includeHeaders: true })) .pipeTo(createCSVWritableStream(new URL('./data/ex5-out.csv', import.meta.url))); \ No newline at end of file diff --git a/examples/ex6.js b/examples/ex6.js index 039de8c..6f77a81 100644 --- a/examples/ex6.js +++ b/examples/ex6.js @@ -1,12 +1,12 @@ // ## Example 6: Handling a flaky transformation (basic retry strategy) // -// Pass a function to the `onError` option of `createCSVTransformStream` to catch errors thrown by the transformation +// Pass a function to the `onError` option of `CSVTransformer` to catch errors thrown by the transformation // function. // // The `onError` function may be used to implement graceful failure strategies (e.g., set the output row to a // placeholder), retry strategies (e.g., a truncated exponential backoff for flaky network requests), logging, etc. -import { createCSVReadableStream, createCSVTransformStream, createCSVWritableStream } from '../src/index.js'; +import { createCSVReadableStream, CSVTransformer, createCSVWritableStream } from '../src/index.js'; // Will immediately retry calling `fn` a maximum of `iterations` times function retry(row, fn, iterations) { @@ -26,5 +26,5 @@ function flaky(row) { } await createCSVReadableStream(new URL('./data/ex6-in.csv', import.meta.url)) - .pipeThrough(createCSVTransformStream(flaky, { onError: (row, e, fn) => retry(row, fn, 1000) })) + .pipeThrough(new CSVTransformer(flaky, { onError: (row, e, fn) => retry(row, fn, 1000) })) .pipeTo(createCSVWritableStream(new URL('./data/ex6-out.csv', import.meta.url))); \ No newline at end of file diff --git a/examples/ex7.js b/examples/ex7.js index 81c4c2b..f70c376 100644 --- a/examples/ex7.js +++ b/examples/ex7.js @@ -4,9 +4,9 @@ // batch to the transformation function. // // This functionality is useful if you need to collect data from multiple rows before processing (e.g., performing an -// expensive network request where is more efficient to batch multiple rows into one request). +// expensive network request where it is more efficient to batch multiple rows into one request). -import { createCSVReadableStream, createCSVTransformStream, createCSVWritableStream } from '../src/index.js'; +import { createCSVReadableStream, CSVTransformer, createCSVWritableStream } from '../src/index.js'; let firstChunk = true; let batchNumber = 0; @@ -29,5 +29,5 @@ function process(batch) { } await createCSVReadableStream(new URL('./data/ex7-in.csv', import.meta.url)) - .pipeThrough(createCSVTransformStream(process, { includeHeaders: true, maxBatchSize: 5 })) + .pipeThrough(new CSVTransformer(process, { includeHeaders: true, maxBatchSize: 5 })) .pipeTo(createCSVWritableStream(new URL('./data/ex7-out.csv', import.meta.url))); \ No newline at end of file diff --git a/src/core.js b/src/core.js index fdb79fd..a86c8ed 100644 --- a/src/core.js +++ b/src/core.js @@ -145,8 +145,8 @@ export function createCSVReadableStream(path) { */ /** - * Options to configure `createCSVTransformStream`. - * @typedef {Object} CreateCSVTransformStreamOptions + * 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`. * @@ -163,92 +163,99 @@ export function createCSVReadableStream(path) { */ /** - * Create a new [TransformStream](https://developer.mozilla.org/en-US/docs/Web/API/TransformStream) to process CSV data. - * - * @param {TransformationFunction} fn A function to process a row or rows of CSV data from `createCSVReadableStream`. - * - * If `options.rawInput` is `false` (default), the input will be a `Array` representing a CSV row. If `options.rawInput` is `true`, - * the input will be a JSON `string` representing a CSV row. - * - * If `options.rawOutput` is `false` (default), the expected output is a `Array` which will be serialized with `JSON.stringify()` and enqueued, - * or a `Array>` which will have each of its sub-arrays serialized and enqueued separately. If `options.rawOutput` is `true`, the raw - * output will be sent the next stream unmodified. - * - * If `options.maxBatchSize` is a number greater than `1`, the input will represent multiple CSV rows as an `Array>` (if - * `options.rawInput` is `false`) or a `string` (if `options.rawInput` is `true`). The input will still be a 2-D array even if there is only - * 1 row in the batch (except for the header row). The header row is always processed by itself and passed as a `Array`, regardless of - * the `options.maxBatchSize` value. - * - * Return `null` to consume an input row without emitting an output row. - * @param {CreateCSVTransformStreamOptions} options Object containing flags to configure the stream logic. - * @returns {TransformStream} A `TransformStream` where each chunk is one row of the CSV file, after transformations applied by `fn()`. + * A `TransformStream` to process CSV data. + * @extends TransformStream */ -export function createCSVTransformStream(fn, options = {}) { - options.includeHeaders ??= false; - options.rawInput ??= false; - options.rawOutput ??= false; - options.onError ??= null; - options.maxBatchSize ??= 1; +export class CSVTransformer extends TransformStream { + #fn; + #options; + #batch = []; + #firstChunk = true; - async function wrappedFn(row) { - if (options.onError === null) { return await fn(row); } - try { return await fn(row); } - catch (e) { return await options.onError(row, e, fn); } + /** + * @param {TransformationFunction} fn A function to process a row or rows of CSV data. + * + * If `options.rawInput` is `false` (default), the input will be a `Array` representing a CSV row. If `options.rawInput` + * is `true`, the input will be a JSON `string` representing a CSV row. + * + * If `options.rawOutput` is `false` (default), the expected output is a `Array` which will be serialized with `JSON.stringify()` + * and enqueued, or a `Array>` which will have each of its sub-arrays serialized and enqueued separately. If + * `options.rawOutput` is `true`, the raw output will be sent to the next stream unmodified. + * + * If `options.maxBatchSize` is a number greater than `1`, the input will represent multiple CSV rows as an `Array>` + * (if `options.rawInput` is `false`) or a `string` (if `options.rawInput` is `true`). The input will still be a 2-D array even + * if there is only 1 row in the batch (except for the header row). The header row is always processed by itself and passed as a + * `Array`, regardless of the `options.maxBatchSize` value. + * + * Return `null` to consume an input row without emitting an output row. + * @param {CSVTransformerOptions} options Object containing flags to configure the stream logic. + */ + constructor(fn, options = {}) { + super({ + transform: (chunk, controller) => this.transform(chunk, controller), + flush: (controller) => this.flush(controller), + }); + + options.includeHeaders ??= false; + options.rawInput ??= false; + options.rawOutput ??= false; + options.onError ??= null; + options.maxBatchSize ??= 1; + + this.#fn = fn; + this.#options = options; } - const batch = []; - let firstChunk = true; - return new TransformStream({ - async transform(chunk, controller) { - const row = options.rawInput ? chunk : JSON.parse(chunk); - let out; - if (firstChunk) { - firstChunk = false; - out = options.includeHeaders ? await wrappedFn(row) : row; - } - else if (options.maxBatchSize > 1) { - batch.push(row); - if (batch.length === options.maxBatchSize) { - out = await wrappedFn(batch); - batch.length = 0; - } - } - else { - out = await wrappedFn(row); - } + async #wrappedFn(row) { + const { onError } = this.#options; + if (onError === null) { return await this.#fn(row); } + try { return await this.#fn(row); } + catch (e) { return await onError(row, e, this.#fn); } + } - if (out === null || out === undefined) { // Input row is consumed without emitting any output row - return; - } + #enqueueRow(row, controller) { + if (row === null || row === undefined) { // Input row is consumed without emitting any output row + return; + } + else if (this.#options.rawOutput || typeof row === 'string') { + controller.enqueue(row); + } + else if (Array.isArray(row) && Array.isArray(row[0])) { // Multiple rows returned, enqueue each row separately + row.forEach(r => controller.enqueue(JSON.stringify(r))); + } + else { + controller.enqueue(JSON.stringify(row)); + } + } - if (options.rawOutput || typeof out === 'string') { - controller.enqueue(out); - } - else if (Array.isArray(out) && Array.isArray(out[0])) { // Multiple rows returned, enqueue each row separately - out.forEach(row => controller.enqueue(JSON.stringify(row))); - } - else { - controller.enqueue(JSON.stringify(out)); - } - }, - async flush(controller) { - if (batch.length > 0) { // CSV ended before batch got filled - const out = await wrappedFn(batch); - if (out === null || out === undefined) { // Input row is consumed without emitting any output row - return; - } - if (options.rawOutput || typeof out === 'string') { - controller.enqueue(out); - } - else if (Array.isArray(out) && Array.isArray(out[0])) { // Multiple rows returned, enqueue each row separately - out.forEach(row => controller.enqueue(JSON.stringify(row))); - } - else { - controller.enqueue(JSON.stringify(out)); - } + async transform(chunk, controller) { + const { includeHeaders, rawInput, maxBatchSize } = this.#options; + const row = rawInput ? chunk : JSON.parse(chunk); + let out; + if (this.#firstChunk) { + this.#firstChunk = false; + out = includeHeaders ? await this.#wrappedFn(row) : row; + } + else if (maxBatchSize > 1) { + this.#batch.push(row); + if (this.#batch.length === maxBatchSize) { + out = await this.#wrappedFn(this.#batch); + this.#batch.length = 0; } } - }); + else { + out = await this.#wrappedFn(row); + } + + this.#enqueueRow(out, controller); + } + + async flush(controller) { + if (this.#batch.length > 0) { + const out = await this.#wrappedFn(this.#batch); + this.#enqueueRow(out, controller); + } + } } /** diff --git a/src/index.js b/src/index.js index a092391..51ebea8 100644 --- a/src/index.js +++ b/src/index.js @@ -1,2 +1,2 @@ -export { createCSVReadableStream, createCSVTransformStream, createCSVWritableStream } from './core.js'; +export { createCSVReadableStream, CSVTransformer, createCSVWritableStream } from './core.js'; export { createCSVNormalizationStream, createCSVDenormalizationStream } from './normalization.js'; diff --git a/test/core.test.js b/test/core.test.js index 3cf8140..9b08754 100644 --- a/test/core.test.js +++ b/test/core.test.js @@ -5,7 +5,7 @@ import { normalize, resolve } from 'node:path'; import { pathToFileURL } from 'node:url'; import { csvStreamEqualWritable, csvStreamNotEqualWritable, createCSVMockStream, createTempFile } from './utils.js'; -import { arrayToCSVString, parsePathLike, createCSVReadableStream, createCSVTransformStream, createCSVWritableStream } from '../src/core.js'; +import { arrayToCSVString, parsePathLike, createCSVReadableStream, CSVTransformer, createCSVWritableStream } from '../src/core.js'; suite('arrayToCSVString', { concurrency: true }, () => { const vectors = [ @@ -191,14 +191,14 @@ suite('createCSVReadableStream', { concurrency: true }, () => { } }); -suite('createCSVTransformStream', { concurrency: true }, () => { +suite('CSVTransformer', { concurrency: true }, () => { test('passes through CSV', { concurrency: true }, async () => { const csv = [ ['columnA', 'columnB'], ['a', 'b'] ]; await createCSVMockStream(csv) - .pipeThrough(createCSVTransformStream(r => r)) + .pipeThrough(new CSVTransformer(r => r)) .pipeTo(csvStreamEqualWritable(csv)); }); test('passes through raw output', { concurrency: true }, async () => { @@ -206,7 +206,7 @@ suite('createCSVTransformStream', { concurrency: true }, () => { ['columnA', 'columnB'], ['a', 'b'] ]) - .pipeThrough(createCSVTransformStream(() => '["abc", "def"]', { includeHeaders: true, rawOutput: true })) + .pipeThrough(new CSVTransformer(() => '["abc", "def"]', { includeHeaders: true, rawOutput: true })) .pipeTo(csvStreamEqualWritable([ ['abc', 'def'], ['abc', 'def'] @@ -223,7 +223,7 @@ suite('createCSVTransformStream', { concurrency: true }, () => { ['a', 'b'], ['should be deleted', 'should be deleted'] ]) - .pipeThrough(createCSVTransformStream(r => r.length % 2 ? r : null)) + .pipeThrough(new CSVTransformer(r => r.length % 2 ? r : null)) .pipeTo(csvStreamEqualWritable([ ['columnA', 'columnB'], ['a', 'b'], @@ -236,7 +236,7 @@ suite('createCSVTransformStream', { concurrency: true }, () => { ['columnA', 'columnB'], ['a', 'b'] ]) - .pipeThrough(createCSVTransformStream(r => ([r, r, r, r]))) + .pipeThrough(new CSVTransformer(r => ([r, r, r, r]))) .pipeTo(csvStreamEqualWritable([ ['columnA', 'columnB'], ['a', 'b'], @@ -251,7 +251,7 @@ suite('createCSVTransformStream', { concurrency: true }, () => { ['1'], ['2'] ]) - .pipeThrough(createCSVTransformStream(() => { throw new Error(); }, { onError: r => [`${r[0]}: caught`] })) + .pipeThrough(new CSVTransformer(() => { throw new Error(); }, { onError: r => [`${r[0]}: caught`] })) .pipeTo(csvStreamEqualWritable([ ['columnA'], ['1: caught'], @@ -264,7 +264,7 @@ suite('createCSVTransformStream', { concurrency: true }, () => { ['columnA'], ['1'] ]) - .pipeThrough(createCSVTransformStream(() => { throw new Error('outer'); }, { onError: () => { throw new Error('inner') } })) + .pipeThrough(new CSVTransformer(() => { throw new Error('outer'); }, { onError: () => { throw new Error('inner') } })) .pipeTo(csvStreamEqualWritable([ ['columnA'], ['1'] @@ -279,7 +279,7 @@ suite('createCSVTransformStream', { concurrency: true }, () => { ['3', '4'], ['5', '6'] ]) - .pipeThrough(createCSVTransformStream(b => b.map(r => r.map(f => Number(f) * 2)), { maxBatchSize: 100 })) + .pipeThrough(new CSVTransformer(b => b.map(r => r.map(f => Number(f) * 2)), { maxBatchSize: 100 })) .pipeTo(csvStreamEqualWritable([ ['columnA', 'columnB'], [2, 4], @@ -294,7 +294,7 @@ suite('createCSVTransformStream', { concurrency: true }, () => { ['3', '4'], ['5', '6'] ]) - .pipeThrough(createCSVTransformStream(b => b.map(r => r.map(f => Number(f) * 2)), { maxBatchSize: 3 })) + .pipeThrough(new CSVTransformer(b => b.map(r => r.map(f => Number(f) * 2)), { maxBatchSize: 3 })) .pipeTo(csvStreamEqualWritable([ ['columnA', 'columnB'], [2, 4], @@ -309,7 +309,7 @@ suite('createCSVTransformStream', { concurrency: true }, () => { ['3', '4'], ['5', '6'] ]) - .pipeThrough(createCSVTransformStream(b => b.map(r => r.map(f => Number(f) * 2)), { maxBatchSize: 2 })) + .pipeThrough(new CSVTransformer(b => b.map(r => r.map(f => Number(f) * 2)), { maxBatchSize: 2 })) .pipeTo(csvStreamEqualWritable([ ['columnA', 'columnB'], [2, 4],