Using classes is more aligned with existing stream APIs such as TextEncoder and CompressionStream.
```javascript
import {
createCSVReadableStream,
- createCSVTransformStream,
+ CSVTransformer,
createCSVWritableStream
} from './src/index.js';
}
await createCSVReadableStream('./data/example-in.csv')
- .pipeThrough(createCSVTransformStream(timesTwo))
+ .pipeThrough(new CSVTransformer(timesTwo))
.pipeTo(createCSVWritableStream('./data/example-out.csv'));
```
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`
// - 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
// - 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) {
}
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
// 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) {
}
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
//
// 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;
}
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
// ## 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) {
}
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
// 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;
}
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
*/
/**
- * 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`.
*
*/
/**
- * 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<string>` 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<any>` which will be serialized with `JSON.stringify()` and enqueued,
- * or a `Array<Array<any>>` 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<Array<string>>` (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<string>`, 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<string>` 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<any>` which will be serialized with `JSON.stringify()`
+ * and enqueued, or a `Array<Array<any>>` 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<Array<string>>`
+ * (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<string>`, 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);
+ }
+ }
}
/**
-export { createCSVReadableStream, createCSVTransformStream, createCSVWritableStream } from './core.js';
+export { createCSVReadableStream, CSVTransformer, createCSVWritableStream } from './core.js';
export { createCSVNormalizationStream, createCSVDenormalizationStream } from './normalization.js';
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 = [
}
});
-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 () => {
['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']
['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'],
['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'],
['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'],
['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']
['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],
['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],
['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],
git.ardis.lu / csvio / commitdiff