Streaming CSV transform with csv-parse + csv-stringify: working versions (tested 2026-10-15)

{
  "name": "csv-parse-stringify-streams",
  "private": true,
  "type": "module",
  "scripts": {
    "test": "node --import tsx --test test/*.test.ts"
  },
  "dependencies": {
    "csv-parse": "7.0.3",
    "csv-stringify": "6.9.0"
  },
  "devDependencies": {
    "tsx": "4.23.15",
    "@types/node": "24.19.2"
  }
}

import { parse } from "csv-parse";
import { stringify } from "csv-stringify";
import { Transform, type Readable, type Writable } from "node:stream";
import { pipeline } from "node:stream/promises";

type Row = { sku: string; qty: string; price: string };

// Adds a total column, drops bad rows. Constant memory for any file size.
export async function addTotals(input: Readable, output: Writable) {
  const stats = { kept: 0, dropped: 0 };
  const logic = new Transform({
    objectMode: true,
    transform(row: Row, _enc, cb) {
      const qty = Number(row.qty), price = Number(row.price);
      if (!qty || Number.isNaN(price)) { stats.dropped++; return cb(); }
      stats.kept++;
      cb(null, { sku: row.sku, qty, price: price.toFixed(2), total: (qty * price).toFixed(2) });
    },
  });
  await pipeline(
    input,
    parse({ columns: true, trim: true, skip_empty_lines: true, bom: true }),
    logic,
    stringify({ header: true, columns: ["sku", "qty", "price", "total"] }),
    output,
  );
  return stats;
}

import { test } from "node:test";
import assert from "node:assert/strict";
import { Readable, Writable } from "node:stream";
import { addTotals } from "../src/transform.ts";

function sink() {
  const chunks: string[] = [];
  return { out: new Writable({ write(c, _e, cb) { chunks.push(c.toString()); cb(); } }), text: () => chunks.join("") };
}

test("adds totals and drops bad rows", async () => {
  const s = sink();
  const stats = await addTotals(Readable.from(["sku,qty,price\n", "A, 2 ,1.5\n", "B,0,9\n", "C,1,abc\n", "\n", "D,3,2\n"]), s.out);
  assert.deepEqual(stats, { kept: 2, dropped: 2 });
  assert.equal(s.text(), "sku,qty,price,total\nA,2,1.50,3.00\nD,3,2.00,6.00\n");
});

test("handles a large generated file as a stream", async () => {
  async function* rows() { yield "sku,qty,price\n"; for (let i = 0; i < 50_000; i++) yield `S${i},1,1\n`; }
  assert.equal((await addTotals(Readable.from(rows()), sink().out)).kept, 50_000);
});

test("malformed CSV rejects the pipeline", async () => {
  await assert.rejects(addTotals(Readable.from(['sku,qty,price\n"A,1,1\n']), sink().out));
});