typescript 119 lines · 3 tabs

Streaming CSV import (Node streams)

Shared by codesnips Jan 2026

3 tabs

import { Writable } from "node:stream";
import type { Pool } from "pg";

interface Row {
  email: string;
  name: string;
  signup_at: string;
}

export class BatchInsertStream extends Writable {
  private buffer: Row[] = [];
  public inserted = 0;

  constructor(private pool: Pool, private batchSize = 500) {
    super({ objectMode: true, highWaterMark: batchSize * 2 });
  }

  async _write(row: Row, _enc: string, callback: (err?: Error) => void) {
    this.buffer.push(row);
    if (this.buffer.length >= this.batchSize) {
      try {
        await this.flush();
      } catch (err) {
        return callback(err as Error);
      }
    }
    callback();
  }

  async _final(callback: (err?: Error) => void) {
    try {
      await this.flush();
      callback();
    } catch (err) {
      callback(err as Error);
    }
  }

  private async flush() {
    if (this.buffer.length === 0) return;
    const rows = this.buffer;
    this.buffer = [];

    const values: unknown[] = [];
    const tuples = rows.map((r, i) => {
      const b = i * 3;
      values.push(r.email, r.name, r.signup_at);
      return `($${b + 1}, $${b + 2}, $${b + 3})`;
    });

    await this.pool.query(
      `INSERT INTO users (email, name, signup_at) VALUES ${tuples.join(",")}
       ON CONFLICT (email) DO NOTHING`,
      values,
    );
    this.inserted += rows.length;
  }
}

import { Transform, type TransformCallback } from "node:stream";

const EMAIL_RE = /^[^@\s]+@[^@\s]+\.[^@\s]+$/;

export class RowValidator extends Transform {
  public rejected = 0;

  constructor(private onReject?: (raw: Record<string, string>, reason: string) => void) {
    super({ objectMode: true });
  }

  _transform(raw: Record<string, string>, _enc: string, cb: TransformCallback) {
    const email = (raw.email ?? "").trim().toLowerCase();
    const name = (raw.name ?? "").trim();

    if (!EMAIL_RE.test(email)) {
      this.rejected++;
      this.onReject?.(raw, "invalid_email");
      return cb();
    }

    const signup = raw.signup_at ? new Date(raw.signup_at) : new Date();
    if (Number.isNaN(signup.getTime())) {
      this.rejected++;
      this.onReject?.(raw, "invalid_date");
      return cb();
    }

    this.push({ email, name, signup_at: signup.toISOString() });
    cb();
  }
}

import { createReadStream } from "node:fs";
import { pipeline } from "node:stream/promises";
import { parse } from "csv-parse";
import type { Pool } from "pg";
import { RowValidator } from "./RowValidator";
import { BatchInsertStream } from "./BatchInsertStream";

export interface ImportResult {
  inserted: number;
  rejected: number;
}

export async function importCsv(path: string, pool: Pool): Promise<ImportResult> {
  const parser = parse({
    columns: (header: string[]) => header.map((h) => h.trim().toLowerCase()),
    skip_empty_lines: true,
    trim: true,
  });

  const validator = new RowValidator((raw, reason) => {
    console.warn(`skipped row (${reason}):`, raw.email ?? "<no email>");
  });

  const writer = new BatchInsertStream(pool, 500);

  await pipeline(createReadStream(path), parser, validator, writer);

  return { inserted: writer.inserted, rejected: validator.rejected };
}

3 files · typescript Explain with highlit

This snippet demonstrates a memory-safe CSV import pipeline built on Node's streaming primitives, the kind of code that ingests a multi-gigabyte export without ever loading the whole file into memory. The core idea is to treat the file as a flow of records rather than a buffer, letting Node's built-in backpressure throttle reads whenever the downstream database writer falls behind.

In BatchInsertStream, a Writable in objectMode accumulates parsed rows into an in-memory buffer and flushes to Postgres only when it reaches batchSize. The crucial detail is the _write callback: it is invoked once per record, and the stream will not pull the next record until callback() fires. By deliberately withholding callback() during a full-batch flush (via await this.flush()), the writer creates natural backpressure that propagates all the way back up the pipe to the file read stream. The _final hook handles the partial last batch, and flush clears the buffer before awaiting the query so a failed insert does not silently re-send stale rows.

The RowValidator tab is a Transform stream that sits between the CSV parser and the writer. It normalizes and validates each object, pushing only clean records downstream and tracking rejected counts instead of throwing, so one bad line does not abort a million-row job. Rejected rows can optionally be diverted to a dead-letter file, a common ETL requirement.

The importCsv orchestrator wires everything together with pipeline from stream/promises. Using pipeline rather than manually chaining .pipe() matters because it guarantees every stream is destroyed and errors propagate correctly even if one stage fails midway, avoiding the classic leaked-file-descriptor and hung-process bugs that plague hand-rolled pipes. The csv-parse library supplies a battle-tested parser configured to emit objects keyed by header.

The trade-off is throughput versus memory: larger batchSize reduces round-trips but raises peak memory and lengthens transaction time. This pattern is the right tool when input size is unbounded or untrusted, when constant memory usage is required, and when partial-failure tolerance beats all-or-nothing correctness.

Related snips

typescript

export interface RetryOptions {
  retries: number;
  baseMs: number;
  maxMs: number;
  signal?: AbortSignal;
  onRetry?: (attempt: number, delay: number, err: unknown) => void;

Exponential backoff with jitter for retries

typescript reliability retry

by codesnips 2 tabs

package dbutil

import (
  "context"

  "github.com/jackc/pgconn"

Retry Postgres serialization failures with bounded attempts

go postgres transactions

by Leah Thompson 1 tab

ruby

require "csv"

class PeopleCsvStream
  include Enumerable

  HEADERS = %w[id full_name email signed_up_at plan].freeze

Resilient CSV Export as a Streamed Response

rails performance streaming

by codesnips 3 tabs

typescript

export type Settled<R> =
  | { status: 'fulfilled'; value: R }
  | { status: 'rejected'; reason: unknown };

export interface ConcurrencyOptions {
  limit: number;

Simple concurrency limiter for batch operations

node concurrency async

by codesnips 2 tabs

typescript

import axios, { AxiosError } from 'axios'
import { v4 as uuidv4 } from 'uuid'

const api = axios.create({
  baseURL: import.meta.env.VITE_API_URL || 'http://localhost:3000/api/v1',
  timeout: 15000,

Axios API client with interceptors

react axios api

by Maya Patel 1 tab

typescript

import { create } from 'zustand'
import { persist } from 'zustand/middleware'

interface FilterState {
  search: string
  category: string | null

Zustand for lightweight state management

react zustand state-management

by Maya Patel 2 tabs

Share this code

Here's the card — post it anywhere.

Streaming CSV import (Node streams) — share card

Link copied