Highland.js #highland #stream #csv

write_to_csv.js

const fs = require('fs')
const path = require('path')
const _H = require('highland')
const csvWriter = require('csv-write-stream')

let filename = path.join(__dirname, 'out.csv')
try { fs.unlinkSync(filename) } catch (err) { }

// https://github.com/maxogden/csv-write-stream
let writer = csvWriter({ headers: ['NAME', 'EMAIL'] })
writer.pipe(fs.createWriteStream(filename))

_H(...)
  // ...
  .map(res => {
    return [res.name, res.email]
  })
  .pipe(writer)

writeCsvRecords.js

import path from "node:path";
import _H from "highland";
import { stringify } from "csv-stringify";
import { createWriteStream } from "node:fs";
import { writeFile } from "node:fs/promises";

/**
 * @param {string} targetFolderPath
 * @param {string} fileName
 * @param {Record<string, unknown>[]} records
 */
export async function writeCsvRecords(targetFolderPath, fileName, records) {
  const filePath = path.join(targetFolderPath, `${fileName}.csv`);
  const recordPick = records[0];

  console.log(`Writing ${fileName}.csv`);

  if (!recordPick) {
    await writeFile(filePath, "", "utf8");
    return;
  }

  const headers = Object.keys(recordPick);

  const stream = _H([headers])
    .concat(_H(records).map((it) => headers.map((h) => String(it[h] ?? ""))))
    .pipe(stringify())
    .pipe(createWriteStream(filePath));

  return await new Promise((resolve, reject) => {
    stream.on("close", resolve);
    stream.on("error", reject);
  });
}

writeCsvRows.js

import _H from "highland";
import { stringify } from "csv-stringify";
import { createWriteStream } from "node:fs";

/**
 * @param {string} filePath
 * @param {unknown[][]} rows
 */
export async function writeCsvRows(filePath, rows) {
  console.log(`Writing ${filePath}`);

  const stream = _H(rows).pipe(stringify()).pipe(createWriteStream(filePath));

  return await new Promise((resolve, reject) => {
    stream.on("close", resolve);
    stream.on("error", reject);
  });
}