```javascript
import {
- createCSVReadableStream,
+ CSVReader,
CSVTransformer,
createCSVWritableStream
} from './src/index.js';
return [Number(row[0]) * 2, Number(row[1]) * 2];
}
-await createCSVReadableStream('./data/example-in.csv')
+await new CSVReader('./data/example-in.csv')
.pipeThrough(new CSVTransformer(timesTwo))
.pipeTo(createCSVWritableStream('./data/example-out.csv'));
```
The core API is contained in `src/core.js` and exposes three functions to read, transform, and write CSV files.
-### `createCSVReadableStream`
+### `CSVReader`
-Create a `ReadableStream` from a local CSV file where each CSV row is one chunk.
+A `ReadableStream` where each chunk is one row from a local CSV file.
```javascript
-import { createCSVReadableStream } from './src/index.js';
+import { CSVReader } from './src/index.js';
-const readableStream = createCSVReadableStream('./data/example-in.csv');
+const readableStream = new CSVReader('./data/example-in.csv');
```
### `CSVTransformer`
// - 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, CSVTransformer, createCSVWritableStream } from '../src/index.js';
+import { CSVReader, 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))
+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
// - 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, CSVTransformer, createCSVWritableStream } from '../src/index.js';
+import { CSVReader, CSVTransformer, createCSVWritableStream } from '../src/index.js';
let firstRow = true;
function sum(row) {
return [...row, Number(row[0]) + Number(row[1])];
}
-await createCSVReadableStream(new URL('./data/ex2-in.csv', import.meta.url))
+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
// transformations (e.g., transformations spread across multiple functions or lower-level transformations directly
// on byte arrays).
-import { createCSVReadableStream, CSVTransformer, createCSVWritableStream } from '../src/index.js';
+import { CSVReader, CSVTransformer, createCSVWritableStream } from '../src/index.js';
// Parse the raw CSV data and transform the data into an object for convenience
function parse(row) {
return [name, paddedId];
}
-await createCSVReadableStream(new URL('./data/ex3-in.csv', import.meta.url))
+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
// `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 { createCSVReadableStream, createCSVWritableStream, createCSVNormalizationStream, createCSVDenormalizationStream } from '../src/index.js';
+import { CSVReader, createCSVWritableStream, 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.
}
];
-await createCSVReadableStream(new URL('./data/ex4-in.csv', import.meta.url))
+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
// ## Example 5: Handling an irregularly-shaped CSV file
//
-// `createCSVReadableStream` can parse CSVs that have rows with a mismatching number of columns, for example CSVs
+// `CSVReader` can parse CSVs that have rows with a mismatching number of columns, for example CSVs
// with informational file headers. This information can be handled as required in the transformation function.
//
// Return `null` in the transformation function to consume an input row without emitting an output row.
-import { createCSVReadableStream, CSVTransformer, createCSVWritableStream } from '../src/index.js';
+import { CSVReader, CSVTransformer, createCSVWritableStream } from '../src/index.js';
let i = 0;
let report, id, date;
return [report, id, date, ...row];
}
-await createCSVReadableStream(new URL('./data/ex5-in.csv', import.meta.url))
+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
// 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, CSVTransformer, createCSVWritableStream } from '../src/index.js';
+import { CSVReader, CSVTransformer, createCSVWritableStream } from '../src/index.js';
// Will immediately retry calling `fn` a maximum of `iterations` times
function retry(row, fn, iterations) {
return [`${v}: pass`];
}
-await createCSVReadableStream(new URL('./data/ex6-in.csv', import.meta.url))
+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
// 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 { createCSVReadableStream, CSVTransformer, createCSVWritableStream } from '../src/index.js';
+import { CSVReader, CSVTransformer, createCSVWritableStream } from '../src/index.js';
let firstChunk = true;
let batchNumber = 0;
return batch.map(row => [batchNumber, sum, ...row]);
}
-await createCSVReadableStream(new URL('./data/ex7-in.csv', import.meta.url))
+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
* Uses Node.js's [`fs.createReadStream()`](https://nodejs.org/api/fs.html#fscreatereadstreampath-options) to read a local
* CSV file then parses the file contents and returns the data through a [Streams API](https://developer.mozilla.org/en-US/docs/Web/API/Streams_API)
* `ReadableStream`.
- * @param {PathLike} path A `string`, `Buffer`, or `URL` representing a path to a local CSV file.
- * @returns {ReadableStream<string>} A `ReadableStream` where each chunk is one row of the CSV file, in the form of a string
- * that may be deserialized using `JSON.parse()`.
+ *
+ * Each chunk is one row of the CSV file, in the form of a string that may be deserialized using `JSON.parse()`.
+ * @extends ReadableStream
*/
-export function createCSVReadableStream(path) {
- const reader = createReadStream(path, { encoding: 'utf-8' });
- const stream = Readable.toWeb(reader);
+export class CSVReader extends ReadableStream {
+ #row = [];
+ #field = '';
+ #lastChar = null;
+ #escapeMode = false;
+ #justExitedEscapeMode = false;
- const row = [];
- let field = '';
- let lastChar = null;
- let escapeMode = false;
- let justExitedEscapeMode = false;
- return stream.pipeThrough(new TransformStream({
- transform(chunk, controller) {
- for (const char of chunk) {
- /* Escape mode logic */
- if (escapeMode) {
- if (char === '"') { // Exit escape mode and do not add char to field
- escapeMode = false;
- justExitedEscapeMode = true;
- }
- else { // Add the literal char because in escape mode
- justExitedEscapeMode = false;
- field += char;
- }
+ /**
+ * @param {PathLike} path A `string`, `Buffer`, or `URL` representing a path to a local CSV file.
+ */
+ constructor(path) {
+ const readStream = createReadStream(path, { encoding: 'utf-8' });
+ const readableStream = Readable.toWeb(readStream).pipeThrough(new TransformStream({
+ transform: (chunk, controller) => this.#transform(chunk, controller),
+ flush: (controller) => this.#flush(controller)
+ }));
+ const reader = readableStream.getReader();
+ super({
+ async pull(controller) {
+ const { done, value } = await reader.read();
+ if (done) {
+ controller.close();
}
- /* Normal mode logic */
else {
- if (char === '"') {
- if (justExitedEscapeMode) { // Re-enter escape mode because the exit char was escaped, and add char to field
- escapeMode = true;
- field += char;
- }
- else if (lastChar === ',' || lastChar === '\n' || lastChar === null) { // Enter escape mode and do not add char to field
- escapeMode = true;
- }
- else { // No special cases match, add the literal char
- field += char;
- }
- }
- else if (char === '\r' || char === '\uFEFF') { } // Ignore CR (newline logic is handled by LF) and byte order mark
- else if (char === ',') { // Terminate the field and do not add char to field
- row.push(field);
- field = '';
+ controller.enqueue(value);
+ }
+ }
+ });
+ }
+
+ #transform(chunk, controller) {
+ for (const char of chunk) {
+ /* Escape mode logic */
+ if (this.#escapeMode) {
+ if (char === '"') { // Exit escape mode and do not add char to field
+ this.#escapeMode = false;
+ this.#justExitedEscapeMode = true;
+ }
+ else { // Add the literal char because in escape mode
+ this.#justExitedEscapeMode = false;
+ this.#field += char;
+ }
+ }
+ /* Normal mode logic */
+ else {
+ if (char === '"') {
+ if (this.#justExitedEscapeMode) { // Re-enter escape mode because the exit char was escaped, and add char to field
+ this.#escapeMode = true;
+ this.#field += char;
}
- else if (char === '\n') { // Terminate the field and row and do not add char to field
- row.push(field);
- field = '';
- controller.enqueue(JSON.stringify(row));
- row.length = 0;
+ else if (this.#lastChar === ',' || this.#lastChar === '\n' || this.#lastChar === null) { // Enter escape mode and do not add char to field
+ this.#escapeMode = true;
}
else { // No special cases match, add the literal char
- field += char;
+ this.#field += char;
}
- justExitedEscapeMode = false;
}
- lastChar = char;
- }
- },
- flush(controller) {
- if (field !== '' || row.length !== 0) { // CSV terminated without a trailing CRLF, leaving a row in the queue
- row.push(field); // Last field is always un-pushed if CSV terminated with nothing
- controller.enqueue(JSON.stringify(row));
- }
- else if (lastChar === null) { // CSV is blank
- controller.enqueue(JSON.stringify(['']));
+ else if (char === '\r' || char === '\uFEFF') { } // Ignore CR (newline logic is handled by LF) and byte order mark
+ else if (char === ',') { // Terminate the field and do not add char to field
+ this.#row.push(this.#field);
+ this.#field = '';
+ }
+ else if (char === '\n') { // Terminate the field and row and do not add char to field
+ this.#row.push(this.#field);
+ this.#field = '';
+ controller.enqueue(JSON.stringify(this.#row));
+ this.#row.length = 0;
+ }
+ else { // No special cases match, add the literal char
+ this.#field += char;
+ }
+ this.#justExitedEscapeMode = false;
}
+ this.#lastChar = char;
+ }
+ }
+
+ #flush(controller) {
+ if (this.#field !== '' || this.#row.length !== 0) { // CSV terminated without a trailing CRLF, leaving a row in the queue
+ this.#row.push(this.#field); // Last field is always un-pushed if CSV terminated with nothing
+ controller.enqueue(JSON.stringify(this.#row));
+ }
+ else if (this.#lastChar === null) { // CSV is blank
+ controller.enqueue(JSON.stringify(['']));
}
- }));
+ }
}
/**
*/
/**
- * A function to process a row or rows of CSV data from `createCSVReadableStream`.
+ * A function to process a row or rows of CSV data from `CSVReader`.
* @callback TransformationFunction
* @param {TransformationInput} row A row or rows of input CSV data before transformation.
* @returns {TransformationOutput} A row or rows of output CSV data after transformation, or null to skip a row.
-export { createCSVReadableStream, CSVTransformer, createCSVWritableStream } from './core.js';
+export { CSVReader, 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, CSVTransformer, createCSVWritableStream } from '../src/core.js';
+import { arrayToCSVString, parsePathLike, CSVReader, CSVTransformer, createCSVWritableStream } from '../src/core.js';
suite('arrayToCSVString', { concurrency: true }, () => {
const vectors = [
}
});
-suite('createCSVReadableStream', { concurrency: true }, () => {
+suite('CSVReader', { concurrency: true }, () => {
const vectors = [
{
name: 'parses simple CSV',
];
for (const { name, input, output, negativeOutput } of vectors) {
test(name, { concurrency: true }, async () => {
- await createCSVReadableStream(input).pipeTo(csvStreamEqualWritable(output));
+ await new CSVReader(input).pipeTo(csvStreamEqualWritable(output));
if (negativeOutput !== undefined) {
- await createCSVReadableStream(input).pipeTo(csvStreamNotEqualWritable(negativeOutput));
+ await new CSVReader(input).pipeTo(csvStreamNotEqualWritable(negativeOutput));
}
});
}
const temp = await createTempFile();
t.after(async () => await unlink(temp));
await createCSVMockStream(csv).pipeTo(createCSVWritableStream(temp));
- await createCSVReadableStream(temp).pipeTo(csvStreamEqualWritable(csv));
+ await new CSVReader(temp).pipeTo(csvStreamEqualWritable(csv));
});
}
});
\ No newline at end of file
import { deepStrictEqual } from 'node:assert/strict';
import { csvStreamEqual, csvStreamEqualWritable, createCSVMockStream } from './utils.js';
-import { createCSVReadableStream } from '../src/core.js';
+import { CSVReader } from '../src/core.js';
import { toCamelCase, expandScientificNotation, fixExcelNumber, fixExcelBigInt, fixExcelDate, createCSVNormalizationStream, createCSVDenormalizationStream } from '../src/normalization.js';
suite('toCamelCase', { concurrency: true }, () => {
{ name: 'bigintCol', displayName: 'BigInt Column', type: 'bigint' },
{ name: 'dateCol', displayName: 'Date Column', type: 'date' }
]
- const stream = createCSVReadableStream('./test/data/normalization.csv')
+ const stream = new CSVReader('./test/data/normalization.csv')
.pipeThrough(createCSVNormalizationStream(headers))
.pipeThrough(createCSVDenormalizationStream());
await csvStreamEqual(stream, [
git.ardis.lu / csvio / commitdiff