faucet_source_rest/config.rs
1//! Stream configuration and builder.
2
3use crate::auth::Auth;
4use crate::pagination::PaginationStyle;
5use faucet_core::AuthSpec;
6use faucet_core::ReplicationMethod;
7use reqwest::{
8 Method,
9 header::{HeaderMap, HeaderName, HeaderValue},
10};
11use schemars::JsonSchema;
12use serde::{Deserialize, Serialize};
13use serde_json::Value;
14use std::collections::HashMap;
15use std::time::Duration;
16
17/// Configuration for a RestStream.
18#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)]
19pub struct RestStreamConfig {
20 // ── Core request ──────────────────────────────────────────────────────────
21 pub base_url: String,
22 /// URL path, relative to `base_url`. May contain `{key}` placeholders that
23 /// are substituted per-partition (e.g. `"/orgs/{org_id}/users"`).
24 pub path: String,
25 #[serde(with = "crate::serde_helpers::http_method")]
26 #[schemars(with = "String")]
27 pub method: Method,
28 /// Authentication: either inline (`{ type, config }`) or a `{ ref: <name> }`
29 /// pointer to a shared provider in the CLI's top-level `auth:` catalog.
30 pub auth: AuthSpec<Auth>,
31 #[serde(skip, default)]
32 pub headers: HeaderMap,
33 pub query_params: HashMap<String, String>,
34 pub body: Option<Value>,
35
36 // ── Pagination ────────────────────────────────────────────────────────────
37 pub pagination: PaginationStyle,
38 pub records_path: Option<String>,
39 pub max_pages: Option<usize>,
40 #[serde(with = "faucet_core::config::duration_secs_option", default)]
41 #[schemars(with = "Option<u64>")]
42 pub request_delay: Option<Duration>,
43
44 // ── Reliability ───────────────────────────────────────────────────────────
45 #[serde(with = "faucet_core::config::duration_secs_option", default)]
46 #[schemars(with = "Option<u64>")]
47 pub timeout: Option<Duration>,
48 /// Number of retries (after the first attempt) for transient request
49 /// failures. Default `3`.
50 ///
51 /// **Precedence note:** the REST source predates the unified pipeline
52 /// `resilience:` policy. When this field (or [`retry_backoff`](Self::retry_backoff))
53 /// is left at its default, an injected `RetryPolicy` (e.g. from a
54 /// pipeline-level `resilience:` block, via
55 /// [`RestStream::with_retry_policy`](crate::RestStream::with_retry_policy))
56 /// governs the retry budget. Setting this field away from its default makes
57 /// it win — an explicit per-connector value is never silently overridden by
58 /// a pipeline-wide default.
59 pub max_retries: u32,
60 /// Base exponential-backoff delay between retries. Default `1s`. Shares the
61 /// legacy-field precedence rule documented on [`max_retries`](Self::max_retries).
62 #[serde(with = "faucet_core::config::duration_secs")]
63 #[schemars(with = "u64")]
64 pub retry_backoff: Duration,
65 /// HTTP status codes that should **not** cause an error. Responses with
66 /// these codes are treated as empty pages (no records, no further pages).
67 pub tolerated_http_errors: Vec<u16>,
68
69 // ── Replication ───────────────────────────────────────────────────────────
70 pub replication_method: ReplicationMethod,
71 /// Field name (not a JSONPath) used for incremental replication bookmarking.
72 pub replication_key: Option<String>,
73 /// Bookmark value: records where `record[replication_key] <= start_replication_value`
74 /// are filtered out when `replication_method` is `Incremental`.
75 pub start_replication_value: Option<Value>,
76 /// Opt-in identifier used by [`Pipeline::with_state_store`](faucet_core::Pipeline::with_state_store)
77 /// to persist this stream's bookmark across runs. When set, the pipeline
78 /// will load any previously-stored bookmark before fetching and write the
79 /// new bookmark only after the sink confirms the batch.
80 ///
81 /// Keys must satisfy [`faucet_core::state::validate_state_key`].
82 pub state_key: Option<String>,
83
84 // ── Singer / Meltano metadata ─────────────────────────────────────────────
85 /// Human-readable stream name (used in logging and Singer SCHEMA messages).
86 pub name: Option<String>,
87 /// Field names that uniquely identify a record (Singer `key_properties`).
88 pub primary_keys: Vec<String>,
89 /// JSON Schema describing the structure of each record.
90 pub schema: Option<Value>,
91 /// Maximum number of records to sample when inferring the schema via
92 /// [`crate::stream::RestStream::infer_schema`]. `0` means sample all
93 /// available records (up to `max_pages`). Defaults to `100`.
94 pub schema_sample_size: usize,
95
96 // ── Partitions ────────────────────────────────────────────────────────────
97 /// Each entry is a context map whose values are substituted into `path`
98 /// placeholders. The stream is executed once per partition and results are
99 /// concatenated. Empty means run once with no substitution.
100 pub partitions: Vec<HashMap<String, Value>>,
101 /// Maximum number of partitions to fetch concurrently.
102 /// `None` means sequential processing (backward compatible default).
103 pub partition_concurrency: Option<usize>,
104}
105
106impl Default for RestStreamConfig {
107 fn default() -> Self {
108 Self {
109 base_url: String::new(),
110 path: String::new(),
111 method: Method::GET,
112 auth: AuthSpec::Inline(Auth::None),
113 headers: HeaderMap::new(),
114 query_params: HashMap::new(),
115 body: None,
116 pagination: PaginationStyle::None,
117 records_path: None,
118 max_pages: Some(100),
119 request_delay: None,
120 timeout: Some(Duration::from_secs(30)),
121 max_retries: 3,
122 retry_backoff: Duration::from_secs(1),
123 tolerated_http_errors: Vec::new(),
124 replication_method: ReplicationMethod::FullTable,
125 replication_key: None,
126 start_replication_value: None,
127 state_key: None,
128 name: None,
129 primary_keys: Vec::new(),
130 schema: None,
131 schema_sample_size: 100,
132 partitions: Vec::new(),
133 partition_concurrency: None,
134 }
135 }
136}
137
138impl RestStreamConfig {
139 pub fn new(base_url: &str, path: &str) -> Self {
140 Self {
141 base_url: base_url.trim_end_matches('/').to_string(),
142 path: path.to_string(),
143 ..Default::default()
144 }
145 }
146
147 // ── Core request ──────────────────────────────────────────────────────────
148
149 pub fn method(mut self, m: Method) -> Self {
150 self.method = m;
151 self
152 }
153
154 pub fn auth(mut self, a: Auth) -> Self {
155 self.auth = AuthSpec::Inline(a);
156 self
157 }
158
159 pub fn header(mut self, k: &str, v: &str) -> Self {
160 self.headers.insert(
161 HeaderName::from_bytes(k.as_bytes()).expect("invalid header name"),
162 HeaderValue::from_str(v).expect("invalid header value"),
163 );
164 self
165 }
166
167 pub fn query(mut self, k: &str, v: &str) -> Self {
168 self.query_params.insert(k.into(), v.into());
169 self
170 }
171
172 pub fn body(mut self, b: Value) -> Self {
173 self.body = Some(b);
174 self
175 }
176
177 // ── Pagination ────────────────────────────────────────────────────────────
178
179 pub fn pagination(mut self, p: PaginationStyle) -> Self {
180 self.pagination = p;
181 self
182 }
183
184 pub fn records_path(mut self, p: &str) -> Self {
185 self.records_path = Some(p.into());
186 self
187 }
188
189 pub fn max_pages(mut self, n: usize) -> Self {
190 self.max_pages = Some(n);
191 self
192 }
193
194 pub fn request_delay(mut self, d: Duration) -> Self {
195 self.request_delay = Some(d);
196 self
197 }
198
199 // ── Reliability ───────────────────────────────────────────────────────────
200
201 pub fn timeout(mut self, d: Duration) -> Self {
202 self.timeout = Some(d);
203 self
204 }
205
206 pub fn max_retries(mut self, n: u32) -> Self {
207 self.max_retries = n;
208 self
209 }
210
211 pub fn retry_backoff(mut self, d: Duration) -> Self {
212 self.retry_backoff = d;
213 self
214 }
215
216 /// HTTP status codes that should be silently ignored (treated as empty pages).
217 pub fn tolerate_http_error(mut self, status: u16) -> Self {
218 self.tolerated_http_errors.push(status);
219 self
220 }
221
222 // ── Replication ───────────────────────────────────────────────────────────
223
224 pub fn replication_method(mut self, m: ReplicationMethod) -> Self {
225 self.replication_method = m;
226 self
227 }
228
229 /// Field name (not JSONPath) used as the incremental replication bookmark.
230 pub fn replication_key(mut self, key: &str) -> Self {
231 self.replication_key = Some(key.into());
232 self
233 }
234
235 /// Bookmark start value: records at or before this value are filtered out
236 /// when using `ReplicationMethod::Incremental`.
237 pub fn start_replication_value(mut self, v: Value) -> Self {
238 self.start_replication_value = Some(v);
239 self
240 }
241
242 /// Opt the stream into resumable runs by giving it a stable state key.
243 /// When this is set and the [`Pipeline`](faucet_core::Pipeline) is
244 /// configured with a state store, the previously persisted bookmark is
245 /// applied to the stream before fetching.
246 pub fn state_key(mut self, key: &str) -> Self {
247 self.state_key = Some(key.into());
248 self
249 }
250
251 // ── Singer / Meltano metadata ─────────────────────────────────────────────
252
253 /// Human-readable stream name.
254 pub fn name(mut self, n: &str) -> Self {
255 self.name = Some(n.into());
256 self
257 }
258
259 /// Field names that uniquely identify a record (Singer `key_properties`).
260 pub fn primary_keys(mut self, keys: Vec<String>) -> Self {
261 self.primary_keys = keys;
262 self
263 }
264
265 /// JSON Schema for the stream's records.
266 pub fn schema(mut self, s: Value) -> Self {
267 self.schema = Some(s);
268 self
269 }
270
271 /// Maximum records to sample for schema inference (`0` = unlimited).
272 pub fn schema_sample_size(mut self, n: usize) -> Self {
273 self.schema_sample_size = n;
274 self
275 }
276
277 // ── Partitions ────────────────────────────────────────────────────────────
278
279 /// Add a partition context. The stream will execute once for each partition,
280 /// substituting `{key}` placeholders in `path` with values from the context.
281 pub fn add_partition(mut self, ctx: HashMap<String, Value>) -> Self {
282 self.partitions.push(ctx);
283 self
284 }
285
286 /// Set the maximum number of partitions to fetch concurrently.
287 /// `None` (default) means sequential processing.
288 pub fn partition_concurrency(mut self, concurrency: Option<usize>) -> Self {
289 self.partition_concurrency = concurrency;
290 self
291 }
292}