faucet-source-rest

A declarative, config-driven REST API client with pluggable authentication, pagination, schema inference, and incremental replication. Attach transforms by wrapping the source with faucet_core::TransformingSource.

Part of the faucet-stream ecosystem.

Installation

[dependencies]
faucet-source-rest = "1.0"
tokio = { version = "1", features = ["full"] }

Or via the umbrella crate:

faucet-stream = { version = "1.0", features = ["source-rest"] }

Quick Start

use faucet_source_rest::{RestStream, RestStreamConfig};

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let config = RestStreamConfig::new("https://api.example.com", "/users")
        .max_pages(5);

    let stream = RestStream::new(config)?;
    let records = stream.fetch_all().await?;

    for record in &records {
        println!("{}", record);
    }
    Ok(())
}

Configuration

Core Request Fields

Pagination Fields

Reliability Fields

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.

Replication Fields

Singer / Meltano Metadata Fields

Partition Fields

Authentication

The Auth enum supports the following strategies:

OAuth2 fields:

• token_url (String): Token endpoint URL
• client_id (String): OAuth2 client ID
• client_secret (String): OAuth2 client secret
• scopes (Vec<String>): Requested scopes
• expiry_ratio (f64): Fraction of expires_in after which the token is refreshed. Must be in (0.0, 1.0]. Defaults to 0.9

TokenEndpoint fields:

• url (String): Token endpoint URL
• method (Method): HTTP method (e.g. POST, GET)
• headers (HeaderMap): Headers for the token request
• body (Option<Value>): Optional JSON body for the token request
• token_path (String): JSONPath expression to extract the token from the response
• expiry_path (Option<String>): JSONPath to extract expiry (seconds) from the response
• expiry_ratio (f64): Proactive refresh ratio. Defaults to 0.9
• response_validator (Option<ResponseValidator>): Custom success check for the token response

Pagination

The PaginationStyle enum supports:

Cursor, LinkHeader, and NextLinkInBody include loop detection -- if the same cursor/link is returned twice in a row, pagination stops. Offset stops on a zero-record page (so a stalled offset cannot loop forever). PageNumber stops when a page returns zero records, and additionally detects content stagnation -- if an API clamps an out-of-range page to the last page and re-returns the identical body, pagination stops rather than looping to max_pages. For all styles, max_pages is a hard cap.

Config Loading

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

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

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

Example JSON config

{
  "base_url": "https://api.github.com",
  "path": "/repos/PawanSikawat/faucet-stream/issues",
  "method": "GET",
  "auth": {
    "type": "bearer",
    "config": {
      "token": "ghp_xxxxxxxxxxxx"
    }
  },
  "query_params": {
    "state": "open",
    "per_page": "100"
  },
  "pagination": {
    "type": "LinkHeader"
  },
  "records_path": "$[*]",
  "max_pages": 10,
  "timeout": 30,
  "max_retries": 3,
  "retry_backoff": 1,
  "tolerated_http_errors": [],
  "replication_method": "FullTable",
  "primary_keys": ["id"],
  "schema_sample_size": 100
}

Example .env file

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

Config Schema Introspection

use faucet_core::Source;

let stream = RestStream::new(config)?;
let schema = stream.config_schema();
println!("{}", serde_json::to_string_pretty(&schema)?);

Methods

RestStream also implements the faucet_core::Source::stream_pages trait method, which is what Pipeline::run drives internally for memory-bounded streaming. The inherent stream_pages() method (which yields Vec<Value> pages) remains for direct callers who do not need per-page bookmarks.

Examples

Cursor-paginated API with bearer auth

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

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)?;
let contacts = stream.fetch_all().await?;

OAuth2 client credentials with incremental replication

use faucet_source_rest::*;
use serde_json::json;

let config = RestStreamConfig::new("https://api.example.com", "/v1/events")
    .auth(Auth::OAuth2 {
        token_url: "https://auth.example.com/oauth/token".into(),
        client_id: "my-client-id".into(),
        client_secret: "my-client-secret".into(),
        scopes: vec!["read:events".into()],
        expiry_ratio: 0.9,
    })
    .pagination(PaginationStyle::Offset {
        offset_param: "offset".into(),
        limit_param: "limit".into(),
        limit: 100,
        total_path: Some("$.total".into()),
    })
    .records_path("$.events[*]")
    .replication_method(ReplicationMethod::Incremental)
    .replication_key("updated_at")
    .start_replication_value(json!("2025-01-01T00:00:00Z"));

let stream = RestStream::new(config)?;
let (records, bookmark) = stream.fetch_all_incremental().await?;
println!("Fetched {} records, new bookmark: {:?}", records.len(), bookmark);

Multi-partition concurrent fetch

use faucet_source_rest::{RestStream, RestStreamConfig, Auth};
use std::collections::HashMap;
use serde_json::json;

let config = RestStreamConfig::new("https://api.example.com", "/orgs/{org_id}/members")
    .auth(Auth::Bearer {
        token: "token".into(),
    })
    .add_partition(HashMap::from([("org_id".into(), json!("acme"))]))
    .add_partition(HashMap::from([("org_id".into(), json!("globex"))]))
    .add_partition(HashMap::from([("org_id".into(), json!("initech"))]))
    .partition_concurrency(Some(3))
    .records_path("$.members[*]");

let stream = RestStream::new(config)?;
let all_members = stream.fetch_all().await?;

Feature Flags

License

Licensed under MIT or Apache-2.0.