exporter

Stream every record of a listing to a CSV or JSON Lines file as pages arrive, or to a writer of your own for Parquet, DuckDB or anything else.

exporter() from outbound-sdk/exporter adds outbound.export(source, { to }). It writes every record of a listing to a .csv or .jsonl file, one row at a time as pages arrive, so a large export never sits in memory. For another format, pass a RowWriter with two functions. The package adds no dependency for Parquet or DuckDB. Writing to a path uses node:fs.

Sample

Runs on the shared seed.

import { mkdtemp, readFile, rm } from "node:fs/promises";
import { tmpdir } from "node:os";
import path from "node:path";

import { createOutbound } from "outbound-sdk";
import { exporter } from "outbound-sdk/exporter";
import { memory } from "outbound-sdk/memory";

import { seed } from "../plugins-index/seed.ts";

const outbound = createOutbound({
  adapter: memory({ data: seed, workspaceId: "agency" }),
  plugins: [exporter()],
});

const dir = await mkdtemp(path.join(tmpdir(), "outbound-export-"));
const csv = path.join(dir, "leads.csv");

// Any listing works as the source. Pages are fetched as rows are written.
const result = await outbound.export(outbound.leads.listAll(), {
  columns: ["ref", "email", "firstName", "company", "custom"],
  to: csv,
});

console.log(result);

console.log(await readFile(csv, "utf-8"));

// JSON Lines, with the default columns: every field of the first record
// except `raw`.
const jsonl = path.join(dir, "campaigns.jsonl");

const campaigns = await outbound.export(outbound.campaigns.listAll(), {
  to: jsonl,
});

console.log(campaigns.columns.join(", "));

const lines = await readFile(jsonl, "utf-8");

console.log(lines.split("\n")[0]);

await rm(dir, { force: true, recursive: true });

{
  columns: [ 'ref', 'email', 'firstName', 'company', 'custom' ],
  rows: 4
}
ref,email,firstName,company,custom
memory:agency:lead_jane,jane.doe@example.com,Jane,Northwind,"{""phone"":""555-0100""}"
memory:agency:lead_sam,sam.lee@example.com,Sam,Contoso,{}
memory:agency:lead_priya,priya.patel@example.com,Priya,Fabrikam,{}
memory:agency:lead_omar,omar.haddad@example.com,Omar,Tailspin,{}

provider, ref, schemaVersion, sourceId, workspaceId, channel, createdAt, name, status, tags
{"provider":"memory","ref":"memory:agency:cmp_founders","schemaVersion":1,"sourceId":"cmp_founders","workspaceId":"agency","channel":"email","createdAt":null,"name":"Acme: SaaS founders","status":"active","tags":["acme"]}

Options

exporter() takes no options.

outbound.export(source, options):

It returns { rows, columns }.

How rows are written

The default columns come from the first record only. A later record with a field the first one lacked does not get a column. Name columns when records vary.

An item of the source that is not an object is skipped and not counted.

A writer of your own

A RowWriter is write(row) and close(), both async. export awaits each write before reading the next record, so a slow writer slows the listing down and does not build a queue.

import { createOutbound, OutboundError } from "outbound-sdk";
import type { Lead } from "outbound-sdk";
import { exporter } from "outbound-sdk/exporter";
import type { RowWriter } from "outbound-sdk/exporter";
import { memory } from "outbound-sdk/memory";

import { seed } from "../plugins-index/seed.ts";

const outbound = createOutbound({
  adapter: memory({ data: seed, workspaceId: "agency" }),
  plugins: [exporter()],
});

// A writer is two async functions. This one collects rows in batches of
// three, the way an insert into DuckDB or a Parquet row group would.
const batchWriter = (label: string): RowWriter => {
  const batches: number[] = [];
  let batch: string[] = [];

  return {
    close: () => {
      if (batch.length > 0) {
        batches.push(batch.length);
      }

      console.log(`${label}: closed after batches of`, batches);

      return Promise.resolve();
    },
    write: (row) => {
      batch.push(String(row.ref));

      if (batch.length === 3) {
        batches.push(batch.length);
        batch = [];
      }

      return Promise.resolve();
    },
  };
};

const result = await outbound.export(outbound.leads.listAll(), {
  to: batchWriter("all leads"),
});

console.log("rows:", result.rows);

// A path that is neither .csv nor .jsonl is refused.
try {
  await outbound.export(outbound.leads.listAll(), { to: "leads.parquet" });
} catch (error) {
  if (OutboundError.is(error)) {
    console.log(`${error.code}: ${error.message}`);
  }
}

// A source that fails part way: the error reaches you, and the writer is
// closed with the rows it already has.
const interrupted = async function* interrupted(): AsyncGenerator<Lead> {
  yield* outbound.leads.listAll({ max: 2 });

  throw new Error("connection lost");
};

try {
  await outbound.export(interrupted(), { to: batchWriter("interrupted") });
} catch (error) {
  console.log("failed:", error instanceof Error ? error.message : error);
}

all leads: closed after batches of [ 3, 1 ]
rows: 4
Invalid: export() writes .csv or .jsonl files; pass a writer for other formats
interrupted: closed after batches of [ 2 ]
failed: connection lost

What it changes

It changes no capability, no record and nothing in meta. It has no layer of its own, so its position in plugins does not matter.

With other plugins

export writes whatever the source yields. A listing from the instance has been through every layer:

• With redact, the file holds masked values and raw is null.
• Without it, the file holds the leads’ addresses and names. It is your file, on your disk. The SDK keeps no copy.
• With clients or identity, the added fields are columns when the first record has them.
• A budget or maxRequests cap stops the listing part way, as below.

When it fails

Limits

• CSV cells are not escaped against spreadsheet formulas. A value that starts with =, +, - or @ is written as it is. Lead fields come from outside your company, so treat an exported CSV as untrusted when opening it in a spreadsheet.
• There is no append mode and no built-in resume. For a copy you keep in step run after run, use sync.