fetch

HTTP client streams with automatic pagination, rate limiting, and 429 retry.

Install

npm install @datastream/fetch

fetchSetDefaults

Set global defaults for all fetch streams. Mutates module-level state — not safe for concurrent multi-tenant use.

Defaults

Option	Type	Default	Description
method	string	"GET"	HTTP method
headers	object	{ Accept: 'application/json', 'Accept-Encoding': 'br, gzip, deflate' }	Request headers
rateLimit	number	0.01	Minimum seconds between requests (0.01 = 100/sec)
dataPath	string	—	Dot-path to data array in JSON response body
nextPath	string	—	Dot-path to next page URL in JSON response body
qs	object	{}	Default query string parameters
offsetParam	string	—	Query parameter name for offset pagination
offsetAmount	number	—	Increment per page for offset pagination

Example

import { fetchSetDefaults } from '@datastream/fetch'

fetchSetDefaults({
  headers: { Authorization: 'Bearer token123' },
  rateLimit: 0.1, // 10 requests/sec
})

fetchReadableStream Readable

Fetches data from one or more URLs and emits chunks. Automatically detects JSON responses and handles pagination.

Also exported as fetchResponseStream.

Options

Option	Type	Default	Description
url	string	—	Request URL
method	string	"GET"	HTTP method
headers	object	—	Request headers (merged with defaults)
rateLimit	number	0.01	Seconds between requests
dataPath	string	—	Dot-path to data in JSON body
nextPath	string	—	Dot-path to next URL in JSON body
qs	object	{}	Query string parameters
offsetParam	string	—	Query param for offset-based pagination
offsetAmount	number	—	Offset increment per page

Pass an array of option objects to fetch from multiple URLs. By default they run one at a time; set concurrency (see below) to fetch in parallel.

Concurrency

Pass concurrency in the second argument (stream options) to fetch array items in parallel. Results are still yielded in array order, and request starts remain spaced by rateLimit. Defaults to 1 (sequential).

fetchReadableStream(
  [
    { url: 'https://api.example.com/a', dataPath: 'data' },
    { url: 'https://api.example.com/b', dataPath: 'data' },
    { url: 'https://api.example.com/c', dataPath: 'data' },
  ],
  { concurrency: 3 },
)

Pagination strategies

Link header — automatically followed when present:

Link: <https://api.example.com/users?page=2>; rel="next"

Body path — use nextPath to extract the next URL from the JSON response:

fetchReadableStream({
  url: 'https://api.example.com/users',
  dataPath: 'data',
  nextPath: 'pagination.next_url',
})

Offset — use offsetParam and offsetAmount for numeric pagination:

fetchReadableStream({
  url: 'https://api.example.com/users',
  dataPath: 'results',
  offsetParam: 'offset',
  offsetAmount: 100,
})

429 retry

When receiving a 429 Too Many Requests response, the request is automatically retried with exponential backoff. If a Retry-After header is present, its value (in seconds) is used as the delay. Otherwise, the delay follows min(1000 × 2^(attempt-1), 30000) ms. Retries are capped at retryMaxCount (default: 10).

Option	Type	Default	Description
retryMaxCount	number	10	Maximum retry attempts on 429

Example

import { pipeline } from '@datastream/core'
import { fetchReadableStream } from '@datastream/fetch'
import { objectCountStream } from '@datastream/object'

const count = objectCountStream()

const result = await pipeline([
  fetchReadableStream({
    url: 'https://api.example.com/users',
    dataPath: 'data',
    nextPath: 'meta.next',
    headers: { Authorization: 'Bearer token' },
  }),
  count,
])

console.log(result)
// { count: 450 }

Multiple URLs

fetchReadableStream([
  { url: 'https://api.example.com/users?status=active', dataPath: 'data' },
  { url: 'https://api.example.com/users?status=inactive', dataPath: 'data' },
])

fetchWritableStream Writable, async

Streams data as the body of an HTTP request. Uses duplex: "half" for browser compatibility.

Also exported as fetchRequestStream.

Example

import { pipeline, createReadableStream } from '@datastream/core'
import { fetchWritableStream } from '@datastream/fetch'
import { csvFormatStream } from '@datastream/csv'

await pipeline([
  createReadableStream(data),
  csvFormatStream({ header: true }),
  await fetchWritableStream({
    url: 'https://api.example.com/upload',
    method: 'PUT',
    headers: { 'Content-Type': 'text/csv' },
  }),
])