import {
CSVReader,
CSVTransformer,
- createCSVWritableStream
+ CSVWriter
} from './src/index.js';
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:
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
// - 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];
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
// - 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) {
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
// 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) {
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
// `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.
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
//
// 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;
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
// 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) {
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
// 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;
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
* 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();
- }
- });
+ });
+ }
}
-export { CSVReader, CSVTransformer, createCSVWritableStream } from './core.js';
+export { CSVReader, CSVTransformer, CSVWriter } 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, CSVReader, CSVTransformer, createCSVWritableStream } from '../src/core.js';
+import { arrayToCSVString, parsePathLike, CSVReader, CSVTransformer, CSVWriter } from '../src/core.js';
suite('arrayToCSVString', { concurrency: true }, () => {
const vectors = [
});
});
-suite('createCSVWritableStream', { concurrency: true }, () => {
+suite('CSVWriter', { concurrency: true }, () => {
const vectors = [
{
name: 'writes simple CSV',
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));
});
}
git.ardis.lu / csvio / commitdiff