node-sf-bulk2

Node.js wrapper for the Salesforce Bulk API 2.0. Works with any auth library that gives you an access token — jsforce, @salesforce/sf-core, SF CLI, etc.

Written in TypeScript with full type definitions included.

Installation

npm install node-sf-bulk2

Quick Start

Create a connection object from any Salesforce auth source, then pass it to BulkAPI2:

import { BulkAPI2, BulkAPI2Connection } from 'node-sf-bulk2';

const connection: BulkAPI2Connection = {
  accessToken: '<your-access-token>',
  instanceUrl: 'https://yourorg.my.salesforce.com',
  apiVersion: '62.0',
};

const bulk = new BulkAPI2(connection);

For the Tooling API, set isTooling: true on the connection. To pass call options, add callOptions:

const connection: BulkAPI2Connection = {
  accessToken: '<token>',
  instanceUrl: 'https://yourorg.my.salesforce.com',
  apiVersion: '62.0',
  isTooling: true,
  callOptions: {
    client: 'myApp',
    defaultNamespace: 'ns1',
  },
};

Query Job — Full Lifecycle

Submit a query, poll until complete, and retrieve CSV results:

import { BulkAPI2, BulkAPI2Connection, QueryInput } from 'node-sf-bulk2';

const connection: BulkAPI2Connection = {
  accessToken: '<token>',
  instanceUrl: 'https://yourorg.my.salesforce.com',
  apiVersion: '62.0',
};
const bulk = new BulkAPI2(connection);

// 1. Submit the query job
const queryInput: QueryInput = {
  query: 'SELECT Id, Name FROM Account',
  operation: 'query',
};
const job = await bulk.submitBulkQueryJob(queryInput);
console.log('Job created:', job.id);

// 2. Poll until complete
let info = await bulk.getBulkQueryJobInfo(job.id);
while (info.state !== 'JobComplete' && info.state !== 'Failed') {
  await new Promise((r) => setTimeout(r, 2000));
  info = await bulk.getBulkQueryJobInfo(job.id);
}

if (info.state === 'Failed') {
  throw new Error('Query job failed');
}

// 3. Get results (CSV)
const response = await bulk.getBulkQueryResults(job.id);
console.log('Results:\n', response.data);

// 4. Clean up
await bulk.deleteBulkQueryJob(job.id);

For large result sets, page through with locator and maxRecords:

let locator: string | undefined;
do {
  const response = await bulk.getBulkQueryResults(job.id, locator, 10000);
  console.log(response.data);
  locator = response.headers['sforce-locator'];
  if (locator === 'null') locator = undefined;
} while (locator);

For large results that shouldn't be buffered in memory, stream directly to a file:

import { createWriteStream } from 'fs';
import { pipeline } from 'stream/promises';

const stream = await bulk.getBulkQueryResultsStream(job.id);
await pipeline(stream, createWriteStream('results.csv'));

Ingest Job — Full Lifecycle

Create an ingest job, upload CSV data, and retrieve results:

import { BulkAPI2, BulkAPI2Connection, JobUploadRequest } from 'node-sf-bulk2';
import { readFileSync } from 'fs';

const connection: BulkAPI2Connection = {
  accessToken: '<token>',
  instanceUrl: 'https://yourorg.my.salesforce.com',
  apiVersion: '62.0',
};
const bulk = new BulkAPI2(connection);

// 1. Create the job
const jobRequest: JobUploadRequest = {
  object: 'Account',
  operation: 'insert',
};
const job = await bulk.createDataUploadJob(jobRequest);

// 2. Upload CSV data
const csv = readFileSync('./accounts.csv', 'utf-8');
await bulk.uploadJobData(job.contentUrl, csv);

// 3. Close the job to begin processing
await bulk.closeOrAbortJob(job.id, 'UploadComplete');

// 4. Poll until complete
let info = await bulk.getIngestJobInfo(job.id);
while (info.state !== 'JobComplete' && info.state !== 'Failed') {
  await new Promise((r) => setTimeout(r, 2000));
  info = await bulk.getIngestJobInfo(job.id);
}

// 5. Get results
const successful = await bulk.getResults(job.id, 0); // successfulResults
const failed = await bulk.getResults(job.id, 1);     // failedResults
console.log('Successful:\n', successful);
console.log('Failed:\n', failed);

// 6. Clean up
await bulk.deleteIngestJob(job.id);

You can also create a job and upload data in a single multipart request:

const csv = readFileSync('./accounts.csv', 'utf-8');
const job = await bulk.createDataUploadJobWithData(
  { object: 'Account', operation: 'insert' },
  csv,
);

API Reference

Query Jobs

| Method | Description | |--------|-------------| | submitBulkQueryJob(query) | Submit a new query job | | getBulkQueryJobInfo(jobId) | Get status/info for a query job | | getAllBulkQueryJobInfo(config?) | List all query jobs (with optional filters) | | getBulkQueryResults(jobId, locator?, maxRecords?) | Get query results as CSV | | getBulkQueryResultsStream(jobId, locator?, maxRecords?) | Stream query results as a Node.js Readable | | getBulkQueryResultPages(jobId) | Get result page URLs for parallel download | | abortBulkQueryJob(jobId) | Abort a query job | | deleteBulkQueryJob(jobId) | Delete a query job |

Ingest Jobs

| Method | Description | |--------|-------------| | createDataUploadJob(request) | Create an ingest job | | createDataUploadJobWithData(request, csv) | Create an ingest job with CSV data in one request | | uploadJobData(contentUrl, csv) | Upload CSV data to an open job | | closeOrAbortJob(jobId, state) | Close (UploadComplete) or abort (Aborted) a job | | getIngestJobInfo(jobId) | Get status/info for an ingest job | | getAllIngestJobInfo(config?) | List all ingest jobs (with optional filters) | | getResults(jobId, resultType) | Get results — 0 = successful, 1 = failed, 2 = unprocessed | | deleteIngestJob(jobId) | Delete an ingest job |

See the full API documentation for details on all request/response types.

Examples

The examples/ directory contains runnable scripts:

• query-job.ts — Submit a query, poll, retrieve CSV results
• ingest-job.ts — Create ingest job, upload CSV, poll, retrieve results

To run them, set your Salesforce credentials as environment variables:

export SF_ACCESS_TOKEN="your-access-token"
export SF_INSTANCE_URL="https://yourorg.my.salesforce.com"

Then run with ts-node:

npx ts-node examples/query-job.ts
npx ts-node examples/ingest-job.ts

Testing

npm test             # run all tests
npm run test:unit    # unit tests only (mocked, no org needed)
npm run test:watch   # watch mode

Integration tests require a Salesforce org:

SF_ACCESS_TOKEN="..." SF_INSTANCE_URL="..." npm run test:integration

License

ISC