From: Ardis Lu Date: Sun, 19 Jan 2025 07:17:16 +0000 (-0800) Subject: Refactor createCSVWritableStream to CSVWriter X-Git-Tag: v0.1.0~115 X-Git-Url: https://git.ardis.lu/?a=commitdiff_plain;h=31569fc5ea3841865c87a9fea88918003b9a3ef9;p=csvio Refactor createCSVWritableStream to CSVWriter --- diff --git a/README.md b/README.md index d625dbf..83d5abe 100644 --- a/README.md +++ b/README.md @@ -21,7 +21,7 @@ Run: import { CSVReader, CSVTransformer, - createCSVWritableStream + CSVWriter } from './src/index.js'; function timesTwo(row) { @@ -30,7 +30,7 @@ function timesTwo(row) { await new CSVReader('./data/example-in.csv') .pipeThrough(new CSVTransformer(timesTwo)) - .pipeTo(createCSVWritableStream('./data/example-out.csv')); + .pipeTo(new CSVWriter('./data/example-out.csv')); ``` A new `./data/example-out.csv` file will be created: @@ -93,14 +93,14 @@ function timesTwo(row) { const transformStream = new CSVTransformer(timesTwo); ``` -### `createCSVWritableStream` +### `CSVWriter` -Create a `WritableStream` to save CSV row data to disk. +A `WritableStream` to save CSV row data to disk. ```javascript -import { createCSVWritableStream } from './src/index.js'; +import { CSVWriter } from './src/index.js'; -const writableStream = createCSVWritableStream('./data/example-out.csv'); +const writableStream = new CSVWriter('./data/example-out.csv'); ``` ## Extended API diff --git a/examples/ex1.js b/examples/ex1.js index 7af1e20..a81f060 100644 --- a/examples/ex1.js +++ b/examples/ex1.js @@ -8,7 +8,7 @@ // - 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 { CSVReader, CSVTransformer, createCSVWritableStream } from '../src/index.js'; +import { CSVReader, CSVTransformer, CSVWriter } from '../src/index.js'; function timesTwo(row) { return [Number(row[0]) * 2, Number(row[1]) * 2]; @@ -16,4 +16,4 @@ function timesTwo(row) { await new CSVReader(new URL('./data/ex1-in.csv', import.meta.url)) .pipeThrough(new CSVTransformer(timesTwo)) - .pipeTo(createCSVWritableStream(new URL('./data/ex1-out.csv', import.meta.url))); \ No newline at end of file + .pipeTo(new CSVWriter(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 25f4c20..de5cdf9 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 { CSVReader, CSVTransformer, createCSVWritableStream } from '../src/index.js'; +import { CSVReader, CSVTransformer, CSVWriter } from '../src/index.js'; let firstRow = true; function sum(row) { @@ -20,4 +20,4 @@ function sum(row) { await new CSVReader(new URL('./data/ex2-in.csv', import.meta.url)) .pipeThrough(new CSVTransformer(sum, { includeHeaders: true })) - .pipeTo(createCSVWritableStream(new URL('./data/ex2-out.csv', import.meta.url))); \ No newline at end of file + .pipeTo(new CSVWriter(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 8e44dc3..42223af 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 { CSVReader, CSVTransformer, createCSVWritableStream } from '../src/index.js'; +import { CSVReader, CSVTransformer, CSVWriter } from '../src/index.js'; // Parse the raw CSV data and transform the data into an object for convenience function parse(row) { @@ -25,4 +25,4 @@ function calculate(obj) { await new CSVReader(new URL('./data/ex3-in.csv', import.meta.url)) .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 + .pipeTo(new CSVWriter(new URL('./data/ex3-out.csv', import.meta.url))); \ No newline at end of file diff --git a/examples/ex4.js b/examples/ex4.js index 59695c9..2d46f86 100644 --- a/examples/ex4.js +++ b/examples/ex4.js @@ -8,7 +8,7 @@ // `createCSVNormalizationStream` performs transformations to un-mangle the above using only a declaration of the header names and data types. // `createCSVDenormalizationStream` converts the data stream back into a CSV row. -import { CSVReader, createCSVWritableStream, createCSVNormalizationStream, createCSVDenormalizationStream } from '../src/index.js'; +import { CSVReader, CSVWriter, createCSVNormalizationStream, createCSVDenormalizationStream } from '../src/index.js'; // The order of items in `headers` is significant, it determines the order of the output CSV columns. // If an input column is not provided in `headers`, the column is removed from the output CSV. @@ -33,4 +33,4 @@ const headers = [ await new CSVReader(new URL('./data/ex4-in.csv', import.meta.url)) .pipeThrough(createCSVNormalizationStream(headers)) .pipeThrough(createCSVDenormalizationStream()) - .pipeTo(createCSVWritableStream(new URL('./data/ex4-out.csv', import.meta.url))); \ No newline at end of file + .pipeTo(new CSVWriter(new URL('./data/ex4-out.csv', import.meta.url))); \ No newline at end of file diff --git a/examples/ex5.js b/examples/ex5.js index 75130fb..d949fcf 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 { CSVReader, CSVTransformer, createCSVWritableStream } from '../src/index.js'; +import { CSVReader, CSVTransformer, CSVWriter } from '../src/index.js'; let i = 0; let report, id, date; @@ -26,4 +26,4 @@ function parse(row) { await new CSVReader(new URL('./data/ex5-in.csv', import.meta.url)) .pipeThrough(new CSVTransformer(parse, { includeHeaders: true })) - .pipeTo(createCSVWritableStream(new URL('./data/ex5-out.csv', import.meta.url))); \ No newline at end of file + .pipeTo(new CSVWriter(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 31aca2f..25b6dfa 100644 --- a/examples/ex6.js +++ b/examples/ex6.js @@ -6,7 +6,7 @@ // 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 { CSVReader, CSVTransformer, createCSVWritableStream } from '../src/index.js'; +import { CSVReader, CSVTransformer, CSVWriter } from '../src/index.js'; // Will immediately retry calling `fn` a maximum of `iterations` times function retry(row, fn, iterations) { @@ -27,4 +27,4 @@ function flaky(row) { await new CSVReader(new URL('./data/ex6-in.csv', import.meta.url)) .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 + .pipeTo(new CSVWriter(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 3f2b52a..21e55a7 100644 --- a/examples/ex7.js +++ b/examples/ex7.js @@ -6,7 +6,7 @@ // This functionality is useful if you need to collect data from multiple rows before processing (e.g., performing an // expensive network request where it is more efficient to batch multiple rows into one request). -import { CSVReader, CSVTransformer, createCSVWritableStream } from '../src/index.js'; +import { CSVReader, CSVTransformer, CSVWriter } from '../src/index.js'; let firstChunk = true; let batchNumber = 0; @@ -30,4 +30,4 @@ function process(batch) { await new CSVReader(new URL('./data/ex7-in.csv', import.meta.url)) .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 + .pipeTo(new CSVWriter(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 da558b6..f9f4762 100644 --- a/src/core.js +++ b/src/core.js @@ -284,32 +284,35 @@ export class CSVTransformer extends TransformStream { * A simple wrapper around Node.js's [`fs.writeFile`](https://nodejs.org/api/fs.html#fspromiseswritefilefile-data-options) to write * streamed data to a file. If the input data is JSON, the data is converted to a string representing a RFC 4180 CSV record. If the * data is not JSON (e.g., if it is already converted to a CSV string), the data is written to the CSV file directly. - * @param {PathLike} path A `string`, `Buffer`, or `URL` representing a path to a local CSV file destination. If the file already - * exists, its **data will be overwritten**. - * @returns {WritableStream} A `WritableStream` to write data to a file on disk. + * @extends WritableStream */ -export function createCSVWritableStream(path) { - const fullPath = parsePathLike(path); - const dir = dirname(fullPath); - let handle; - - return new WritableStream({ - async start() { - await mkdir(dir, { recursive: true }); - handle = await open(fullPath, 'w'); - }, - async write(chunk) { - let data; - try { - data = arrayToCSVString(JSON.parse(chunk)); - } - catch { - data = chunk; +export class CSVWriter extends WritableStream { + /** + * @param {PathLike} path A `string`, `Buffer`, or `URL` representing a path to a local CSV file destination. If the file already + * exists, its **data will be overwritten**. + */ + constructor(path) { + const fullPath = parsePathLike(path); + const dir = dirname(fullPath); + let handle; + super({ + async start() { + await mkdir(dir, { recursive: true }); + handle = await open(fullPath, 'w'); + }, + async write(chunk) { + let data; + try { + data = arrayToCSVString(JSON.parse(chunk)); + } + catch { + data = chunk; + } + await handle.write(data); + }, + async close() { + await handle.close(); } - await handle.write(data); - }, - async close() { - await handle.close(); - } - }); + }); + } } diff --git a/src/index.js b/src/index.js index c75b66b..c78b4f1 100644 --- a/src/index.js +++ b/src/index.js @@ -1,2 +1,2 @@ -export { CSVReader, CSVTransformer, createCSVWritableStream } from './core.js'; +export { CSVReader, CSVTransformer, CSVWriter } from './core.js'; export { createCSVNormalizationStream, createCSVDenormalizationStream } from './normalization.js'; diff --git a/test/core.test.js b/test/core.test.js index e850c27..eddc5d1 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, CSVReader, CSVTransformer, createCSVWritableStream } from '../src/core.js'; +import { arrayToCSVString, parsePathLike, CSVReader, CSVTransformer, CSVWriter } from '../src/core.js'; suite('arrayToCSVString', { concurrency: true }, () => { const vectors = [ @@ -319,7 +319,7 @@ suite('CSVTransformer', { concurrency: true }, () => { }); }); -suite('createCSVWritableStream', { concurrency: true }, () => { +suite('CSVWriter', { concurrency: true }, () => { const vectors = [ { name: 'writes simple CSV', @@ -421,7 +421,7 @@ suite('createCSVWritableStream', { concurrency: true }, () => { test(name, { concurrency: true }, async (t) => { const temp = await createTempFile(); t.after(async () => await unlink(temp)); - await createCSVMockStream(csv).pipeTo(createCSVWritableStream(temp)); + await createCSVMockStream(csv).pipeTo(new CSVWriter(temp)); await new CSVReader(temp).pipeTo(csvStreamEqualWritable(csv)); }); }