faucet-source-rest

A declarative, config-driven REST API source with pluggable authentication, six pagination styles, schema inference, and incremental replication. Part of the faucet-stream ecosystem.

This is the flagship faucet-stream source: point it at any JSON-over-HTTP API, describe how to authenticate and paginate, and it streams every record page-by-page into any faucet-stream sink — with retries, Retry-After honouring, loop detection, and resumable bookmarks — all from one YAML config and no glue code.

Feature highlights

• Six pagination styles — Cursor, LinkHeader, NextLinkInBody, PageNumber, Offset, and None, each with its own termination/loop guard so a misbehaving API can't loop forever.
• Eight auth methods — bearer, basic, api_key (header), api_key_query, oauth2 (client credentials with token caching), token_endpoint (fetch a token from an arbitrary endpoint), custom headers, and none — plus shared auth: { ref } providers via the CLI's top-level auth: catalog.
• Memory-bounded streaming — overrides Source::stream_pages, so Pipeline::run writes each page to the sink as it arrives; peak memory stays O(page) regardless of total record count.
• Resilient by default — exponential backoff with jitter (capped at 60 s), 429 Retry-After (delta-seconds or HTTP-date) honouring, and a tolerated_http_errors allowlist for legitimately-absent resources.
• Incremental replication — bookmark by any record field, persist it across runs with a state store, and resume from the last value.
• Concurrent partitions — fan a single config across many path substitutions (/orgs/{org_id}/users) and fetch them concurrently.
• Schema inference — sample records to produce a JSON Schema, or supply your own; Singer/Meltano primary_keys / name metadata is carried through.
• Client built once — the reqwest client is constructed in new() and reused for every request and partition.

Installation

# As a library:
cargo add faucet-source-rest
cargo add tokio --features full

# In the CLI (source-rest is a DEFAULT feature — already enabled):
cargo install faucet-cli

The umbrella crate enables it by default too:

cargo add faucet-stream --features source-rest

Quick start

# pipeline.yaml — faucet run pipeline.yaml
version: 1
name: github_issues_to_jsonl

pipeline:
  source:
    type: rest
    config:
      base_url: https://api.github.com
      path: /repos/PawanSikawat/faucet-stream/issues
      method: GET
      auth:
        type: bearer
        config:
          token: ${env:GITHUB_TOKEN}
      query_params:
        state: open
        per_page: "100"
      pagination:
        type: LinkHeader
  sink:
    type: jsonl
    config:
      path: ./out/issues.jsonl

faucet run pipeline.yaml

Configuration reference

Core request

Pagination

Reliability

A 204 No Content response — or any 2xx with an empty/whitespace-only body — is treated as an empty page ("no data"), not a parse error. A non-empty body that isn't valid JSON still fails loudly with FaucetError::Json.

Unified resilience: policy

When driven by the CLI, a pipeline-level resilience: block can inject one shared retry policy into this source. Legacy fields win when set explicitly: if you set max_retries or retry_backoff to anything other than their defaults (3 / 1), the per-connector value is used and the injected policy is ignored for that field — an explicit setting is never silently overridden. Otherwise the injected policy applies.

Because the REST source keeps its own 429/Retry-After-aware retry runner, it honors only the injected policy's max_attempts (→ max_retries) and base (→ retry_backoff). The policy's retry_on, max (per-sleep cap), and jitter fields are inert on REST — they are honored on the xml/graphql sources and on every sink-side write.

Replication & state

Singer / Meltano metadata

Partitions

Authentication

The auth field accepts the project-wide adjacently-tagged { type, config } shape (snake_case discriminators), or a { ref: <name> } pointer into the CLI's top-level shared auth: catalog.

oauth2 / token_endpoint notes: expiry_ratio is the fraction of the token lifetime after which the cached token is proactively refreshed — must be in (0.0, 1.0], defaults to 0.9. For token_endpoint, token_path is the JSONPath to the token string and expiry_path (optional) is the JSONPath to the expiry in seconds; when absent the token is cached indefinitely. A cached token that the API later rejects with 401 Unauthorized (a server-side expiry the time-based cache can't see, including the cached-indefinitely case) is invalidated and the request is retried once with a freshly-fetched token, so a long run doesn't abort mid-way.

# Bearer
auth:
  type: bearer
  config:
    token: ${env:GITHUB_TOKEN}

# API key header
auth:
  type: api_key
  config:
    header: X-API-Key
    value: ${env:API_KEY}

# OAuth2 client credentials
auth:
  type: oauth2
  config:
    token_url: https://auth.example.com/oauth/token
    client_id: ${env:CLIENT_ID}
    client_secret: ${env:CLIENT_SECRET}
    scopes: ["read:events"]
    expiry_ratio: 0.9

# Shared provider from the top-level auth: catalog
auth: { ref: my_idp }

Examples

Cursor-paginated API with bearer auth

source:
  type: rest
  config:
    base_url: https://api.example.com
    path: /v2/contacts
    auth:
      type: bearer
      config:
        token: ${env:API_TOKEN}
    pagination:
      type: Cursor
      next_token_path: $.meta.next_cursor
      param_name: cursor
    records_path: $.data[*]
    max_pages: 50

OAuth2 + incremental replication with a persisted bookmark

version: 1
name: events_incremental
pipeline:
  source:
    type: rest
    config:
      base_url: https://api.example.com
      path: /v1/events
      auth:
        type: oauth2
        config:
          token_url: https://auth.example.com/oauth/token
          client_id: ${env:CLIENT_ID}
          client_secret: ${env:CLIENT_SECRET}
          scopes: ["read:events"]
          expiry_ratio: 0.9
      pagination:
        type: Offset
        offset_param: offset
        limit_param: limit
        limit: 100
        total_path: $.total
      records_path: $.events[*]
      replication_method:
        type: Incremental
      replication_key: updated_at
      start_replication_value: "2026-01-01T00:00:00Z"
      state_key: events_stream
  sink:
    type: jsonl
    config:
      path: ./out/events.jsonl
  state:
    type: file
    config:
      path: ./state.json

Multi-partition concurrent fetch

source:
  type: rest
  config:
    base_url: https://api.example.com
    path: /orgs/{org_id}/members
    auth:
      type: bearer
      config:
        token: ${env:API_TOKEN}
    records_path: $.members[*]
    partitions:
      - { org_id: acme }
      - { org_id: globex }
      - { org_id: initech }
    partition_concurrency: 3

Tolerating a 429 and capping retries

source:
  type: rest
  config:
    base_url: https://api.example.com
    path: /v1/comments
    auth:
      type: api_key
      config:
        header: X-API-Key
        value: ${env:API_KEY}
    pagination:
      type: LinkHeader
    records_path: $.items[*]
    timeout: 60
    max_retries: 5
    retry_backoff: 2
    tolerated_http_errors: [429]

Pagination

The pagination field selects a PaginationStyle (tagged by type). max_pages is a hard cap across all styles.

Streaming & batching

RestStream overrides Source::stream_pages, fetching the next HTTP page on demand and yielding it as a StreamPage. Pipeline::run writes each page to the sink as it arrives, so peak memory is bounded at one page no matter how large the feed. The bookmark is carried on the final page (incremental mode) so the state store advances only after the sink confirms the full run.

The inherent stream_pages() method (yielding Vec<Value> pages, no per-page bookmark) remains for direct callers, alongside the eager fetch_all() / fetch_all_incremental() helpers.

Resume & state

This source supports resumable runs. Set state_key and configure a state: block (or call Pipeline::with_state_store from Rust). On each run the pipeline:

1. loads the previously persisted bookmark and applies it via apply_start_bookmark (overriding start_replication_value);
2. fetches only records newer than the bookmark (replication_method: Incremental + replication_key);
3. persists the new bookmark only after the sink confirms the batch — so a crash mid-run re-fetches rather than skips.

state_key must satisfy faucet_core::state::validate_state_key. See the second example above.

Config loading & schema introspection

Configs load from YAML/JSON files or environment variables:

use faucet_core::config::{load_json, load_env_file};
use faucet_source_rest::RestStreamConfig;

// From a JSON file:
let config: RestStreamConfig = load_json("config.json")?;

// From a .env file + environment (REST_ prefix):
let config: RestStreamConfig = load_env_file(".env", "REST")?;

Example .env:

REST_BASE_URL=https://api.github.com
REST_PATH=/repos/PawanSikawat/faucet-stream/issues
REST_METHOD=GET
REST_MAX_PAGES=10
REST_TIMEOUT=30
REST_MAX_RETRIES=3
REST_RETRY_BACKOFF=1
REST_SCHEMA_SAMPLE_SIZE=100

Inspect the full JSON Schema with:

faucet schema source rest

Library usage

use faucet_source_rest::{RestStream, RestStreamConfig, Auth, PaginationStyle};

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let config = RestStreamConfig::new("https://api.example.com", "/v2/contacts")
        .auth(Auth::Bearer { token: "your-api-token".into() })
        .pagination(PaginationStyle::Cursor {
            next_token_path: "$.meta.next_cursor".into(),
            param_name: "cursor".into(),
        })
        .records_path("$.data[*]")
        .max_pages(50);

    let stream = RestStream::new(config)?;          // validates auth at construction
    let contacts = stream.fetch_all().await?;
    println!("fetched {} records", contacts.len());
    Ok(())
}

Inherent helper methods on RestStream:

Attach transforms by wrapping the source with faucet_core::TransformingSource.

How it works

1. new() resolves the auth method and builds the reqwest client once, reusing it for every request and partition.
2. For each partition, {key} placeholders in path are substituted from the context map; with partition_concurrency set, partitions run concurrently.
3. Each page request is wrapped in the retry layer: transient failures back off exponentially with jitter (capped at 60 s); 429 honours Retry-After.
4. Records are extracted from the response body via records_path (JSONPath); the pagination style decides the next request and when to stop.
5. In Incremental mode, records at or before the bookmark are filtered out and the max replication-key value becomes the new bookmark, carried on the final page.

Lineage dataset URI

https://<base_url><path> (credentials stripped) — e.g. https://api.example.com/v1/users.

Feature flags

Troubleshooting / FAQ

See also

• Connector catalog & capability matrix
• Authentication cookbook
• Pagination cookbook
• Resumable state & bookmarks
• Config-file grammar
• Related crates: faucet-source-graphql, faucet-source-xml, faucet-sink-http, faucet-auth.

License

Licensed under either of Apache License, Version 2.0 or MIT license at your option.