Struct RestStream 

pub struct RestStream { /* private fields */ }

A configured REST API stream that handles pagination, auth, and extraction.

Implementations

impl RestStream

pub fn new(config: RestStreamConfig) -> Result<Self, FaucetError>

Create a new stream from the given configuration.

pub fn with_auth_provider(self, provider: SharedAuthProvider) -> Self

Attach a shared AuthProvider. When set, the provider supplies the credential for every request (taking precedence over inline auth), so several sources can share one token with single-flight refresh. Used by the CLI to resolve auth: { ref }, and by library callers who construct one provider and inject it into many sources.

pub async fn fetch_all(&self) -> Result<Vec<Value>, FaucetError>

Fetch all records across all pages as raw JSON values.

When partitions are configured, the stream is executed once per partition and all results are concatenated.

When replication_method is Incremental and replication_key + start_replication_value are both set, records at or before the bookmark are filtered out.

pub async fn fetch_all_as<T: for<'de> Deserialize<'de>>( &self, ) -> Result<Vec<T>, FaucetError>

Fetch all records and deserialize into typed structs.

pub async fn infer_schema(&self) -> Result<Value, FaucetError>

Infer a JSON Schema for this stream’s records.

If a schema is already set on the config, it is returned immediately without making any HTTP requests.

Otherwise the stream fetches up to schema_sample_size records (respecting max_pages) and derives a JSON Schema from them. Fields that are absent in some records, or that carry a null value, are marked as nullable (["<type>", "null"]).

Set schema_sample_size to 0 to sample all available records.

pub async fn fetch_all_incremental( &self, ) -> Result<(Vec<Value>, Option<Value>), FaucetError>

Fetch all records in incremental mode, returning the records along with the maximum value of replication_key observed across those records.

The returned bookmark should be persisted by the caller and passed back as start_replication_value on the next run.

If no replication_key is configured, this behaves identically to fetch_all and the bookmark is None.

pub fn stream_pages( &self, ) -> Pin<Box<dyn Stream<Item = Result<Vec<Value>, FaucetError>> + Send + '_>>

Stream API pages without buffering the full result set.

This is a thin convenience wrapper around the Source::stream_pages trait method — it discards bookmarks and yields one Vec<Value> per upstream API page. Use the trait method directly if you need per-page bookmarks for incremental replication.

Note: partitions are not supported by stream_pages. Use fetch_all for multi-partition streams.

use faucet_source_rest::{RestStream, RestStreamConfig};
use futures::StreamExt;

let stream = RestStream::new(RestStreamConfig::new("https://api.example.com", "/items"))?;
let mut pages = stream.stream_pages();
while let Some(page) = pages.next().await {
    let records = page?;
    println!("got {} records", records.len());
}

Trait Implementations

impl Source for RestStream

fn fetch_with_context<'life0, 'life1, 'async_trait>( &'life0 self, context: &'life1 HashMap<String, Value>, ) -> Pin<Box<dyn Future<Output = Result<Vec<Value>, FaucetError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Primary fetch method. Receives context from a parent source’s records.

fn fetch_with_context_incremental<'life0, 'life1, 'async_trait>( &'life0 self, context: &'life1 HashMap<String, Value>, ) -> Pin<Box<dyn Future<Output = Result<(Vec<Value>, Option<Value>), FaucetError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Incremental fetch with parent context support.

fn connector_name(&self) -> &'static str

Stable identifier used as the connector label on metrics and the connector attribute on spans. Defaults to the final segment of std::any::type_name::<Self>(), e.g. "RestSource". Built-in connectors override with a short, friendly snake_case name (e.g. "rest"). Must return a non-empty string; observability decorators fall back to "unknown" in release builds if it is empty (and debug_assert! in debug builds).

fn config_schema(&self) -> Value

Return a JSON Schema describing the configuration this source accepts.

fn state_key(&self) -> Option<String>

Stable key under which this source’s incremental-replication bookmark should be persisted in a StateStore.

fn stream_pages<'a>( &'a self, context: &'a HashMap<String, Value>, _batch_size: usize, ) -> Pin<Box<dyn Stream<Item = Result<StreamPage, FaucetError>> + Send + 'a>>

Stream records page-by-page so the pipeline can write to the sink as pages arrive instead of buffering the full result set.

fn apply_start_bookmark<'life0, 'async_trait>( &'life0 self, bookmark: Value, ) -> Pin<Box<dyn Future<Output = Result<(), FaucetError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Apply a bookmark loaded from a StateStore as this run’s starting point.

fn fetch_all<'life0, 'async_trait>( &'life0 self, ) -> Pin<Box<dyn Future<Output = Result<Vec<Value>, FaucetError>> + Send + 'async_trait>>

where 'life0: 'async_trait, Self: 'async_trait,

Convenience: fetch with no parent context.

fn fetch_all_incremental<'life0, 'async_trait>( &'life0 self, ) -> Pin<Box<dyn Future<Output = Result<(Vec<Value>, Option<Value>), FaucetError>> + Send + 'async_trait>>

where 'life0: 'async_trait, Self: 'async_trait,

Convenience: incremental fetch with no parent context.

fn check<'life0, 'life1, 'async_trait>( &'life0 self, ctx: &'life1 CheckContext, ) -> Pin<Box<dyn Future<Output = Result<CheckReport, FaucetError>> + Send + 'async_trait>>

where 'life0: 'async_trait, 'life1: 'async_trait, Self: 'async_trait,

Run a fast, non-mutating preflight probe (used by faucet doctor).