qubit_http/request/
http_request_builder.rs

1/*******************************************************************************
2 *
3 *    Copyright (c) 2025 - 2026 Haixing Hu.
4 *
5 *    SPDX-License-Identifier: Apache-2.0
6 *
7 *    Licensed under the Apache License, Version 2.0.
8 *
9 ******************************************************************************/
10//! Builder for [`super::http_request::HttpRequest`].
11
12use std::time::Duration;
13use std::{
14    fmt,
15    future::Future,
16    pin::Pin,
17};
18
19use bytes::Bytes;
20use http::header::CONTENT_TYPE;
21use http::{
22    HeaderMap,
23    HeaderValue,
24    Method,
25};
26use serde::Serialize;
27use tokio_util::sync::CancellationToken;
28use url::form_urlencoded;
29use url::Url;
30
31use crate::content_type;
32use crate::sanitize::SanitizedDebugger;
33use crate::{
34    AsyncHttpHeaderInjector,
35    HttpClient,
36    HttpError,
37    HttpHeaderInjector,
38    HttpRequestBodyByteStream,
39    HttpRequestStreamingBody,
40    HttpResult,
41    HttpRetryMethodPolicy,
42    LogSanitizePolicy,
43};
44
45use super::http_request::HttpRequest;
46use super::http_request_body::HttpRequestBody;
47use super::http_request_retry_override::HttpRequestRetryOverride;
48use super::parse_header;
49use super::validate_positive_timeout;
50
51/// Builder for [`HttpRequest`](super::http_request::HttpRequest).
52#[derive(Clone)]
53pub struct HttpRequestBuilder {
54    /// HTTP method (e.g. GET, POST).
55    pub(super) method: Method,
56    /// Request path without the query string.
57    pub(super) path: String,
58    /// Query parameters as `(key, value)` pairs, appended to the URL when built.
59    pub(super) query: Vec<(String, String)>,
60    /// Request headers.
61    pub(super) headers: HeaderMap,
62    /// Request body; empty if not set.
63    pub(super) body: HttpRequestBody,
64    /// Deferred streaming upload body factory for per-attempt stream creation.
65    pub(super) streaming_body: Option<HttpRequestStreamingBody>,
66    /// Per-request timeout; if unset, the client default applies.
67    pub(super) request_timeout: Option<Duration>,
68    /// Per-request write timeout used by the send phase.
69    pub(super) write_timeout: Duration,
70    /// Per-request read timeout used by buffered/stream response reading.
71    pub(super) read_timeout: Duration,
72    /// Base URL copied from client options and used by [`HttpRequest::resolved_url`].
73    pub(super) base_url: Option<Url>,
74    /// Whether IPv6 literal hosts are rejected during URL resolution.
75    pub(super) ipv4_only: bool,
76    /// Optional cancellation token for this request.
77    pub(super) cancellation_token: Option<CancellationToken>,
78    /// Per-request retry override for one-off retry behavior customization.
79    pub(super) retry_override: HttpRequestRetryOverride,
80    /// Default headers snapshot from the originating client.
81    pub(super) default_headers: HeaderMap,
82    /// Sync header injectors snapshot from the originating client.
83    pub(super) injectors: Vec<HttpHeaderInjector>,
84    /// Async header injectors snapshot from the originating client.
85    pub(super) async_injectors: Vec<AsyncHttpHeaderInjector>,
86    /// Log sanitization policy snapshot from the originating client.
87    pub(super) log_sanitize_policy: LogSanitizePolicy,
88}
89
90impl fmt::Debug for HttpRequestBuilder {
91    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
92        let debugger = SanitizedDebugger::new(&self.log_sanitize_policy);
93        let url = self.debug_resolved_url().map(|url| debugger.url(&url));
94        let base_url = self.base_url.as_ref().map(|url| debugger.url(url));
95        formatter
96            .debug_struct("HttpRequestBuilder")
97            .field("method", &self.method)
98            .field("url", &url)
99            .field("headers", &debugger.headers(&self.headers))
100            .field("body", &self.body)
101            .field("streaming_body", &self.streaming_body.as_ref().map(|_| "present"))
102            .field("request_timeout", &self.request_timeout)
103            .field("write_timeout", &self.write_timeout)
104            .field("read_timeout", &self.read_timeout)
105            .field("base_url", &base_url)
106            .field("ipv4_only", &self.ipv4_only)
107            .field("cancellation_token_present", &self.cancellation_token.is_some())
108            .field("retry_override", &self.retry_override)
109            .field("default_headers", &debugger.headers(&self.default_headers))
110            .field("injector_count", &self.injectors.len())
111            .field("async_injector_count", &self.async_injectors.len())
112            .finish()
113    }
114}
115
116impl HttpRequestBuilder {
117    /// Starts a builder with method/path and copies supported defaults from client options.
118    ///
119    /// # Parameters
120    /// - `method`: HTTP verb.
121    /// - `path`: URL or relative path string.
122    /// - `client`: Source client whose relevant defaults are copied into this builder.
123    ///
124    /// # Returns
125    /// New [`HttpRequestBuilder`].
126    pub(crate) fn new(method: Method, path: &str, client: &HttpClient) -> Self {
127        let options = client.options();
128        Self {
129            method,
130            path: path.to_string(),
131            query: Vec::new(),
132            headers: HeaderMap::new(),
133            body: HttpRequestBody::Empty,
134            streaming_body: None,
135            request_timeout: options.timeouts.request_timeout,
136            write_timeout: options.timeouts.write_timeout,
137            read_timeout: options.timeouts.read_timeout,
138            base_url: options.base_url.clone(),
139            ipv4_only: options.ipv4_only,
140            cancellation_token: None,
141            retry_override: HttpRequestRetryOverride::default(),
142            default_headers: client.headers_snapshot(),
143            injectors: client.injectors_snapshot(),
144            async_injectors: client.async_injectors_snapshot(),
145            log_sanitize_policy: options.log_sanitize_policy.clone(),
146        }
147    }
148
149    /// Appends a single `key=value` query pair (order preserved).
150    ///
151    /// # Parameters
152    /// - `key`: Query parameter name.
153    /// - `value`: Query parameter value.
154    ///
155    /// # Returns
156    /// `self` for chaining.
157    pub fn query_param(mut self, key: &str, value: &str) -> Self {
158        self.query.push((key.to_string(), value.to_string()));
159        self
160    }
161
162    /// Appends many query pairs via [`HttpRequestBuilder::query_param`].
163    ///
164    /// # Parameters
165    /// - `params`: Iterator of `(key, value)` pairs.
166    ///
167    /// # Returns
168    /// `self` for chaining.
169    pub fn query_params<'a, I>(mut self, params: I) -> Self
170    where
171        I: IntoIterator<Item = (&'a str, &'a str)>,
172    {
173        for (key, value) in params {
174            self = self.query_param(key, value);
175        }
176        self
177    }
178
179    /// Validates and inserts one header.
180    ///
181    /// # Parameters
182    /// - `name`: Header name (must be valid [`http::header::HeaderName`] bytes).
183    /// - `value`: Header value (must be valid [`http::header::HeaderValue`]).
184    ///
185    /// # Returns
186    /// `Ok(self)` or [`HttpError`] if name/value are invalid.
187    pub fn header(mut self, name: &str, value: &str) -> HttpResult<Self> {
188        let (header_name, header_value) = parse_header(name, value)?;
189        self.headers.insert(header_name, header_value);
190        Ok(self)
191    }
192
193    /// Merges all entries from `headers` into this builder (existing names may get extra values).
194    ///
195    /// # Parameters
196    /// - `headers`: Map to append.
197    ///
198    /// # Returns
199    /// `self` for chaining.
200    pub fn headers(mut self, headers: HeaderMap) -> Self {
201        self.headers.extend(headers);
202        self
203    }
204
205    /// Sets the body to raw bytes without changing `Content-Type` unless already set elsewhere.
206    ///
207    /// # Parameters
208    /// - `body`: Payload.
209    ///
210    /// # Returns
211    /// `self` for chaining.
212    pub fn bytes_body(mut self, body: impl Into<Bytes>) -> Self {
213        self.body = HttpRequestBody::Bytes(body.into());
214        self.streaming_body = None;
215        self
216    }
217
218    /// Sets the body to an ordered chunk stream for incremental upload.
219    ///
220    /// # Parameters
221    /// - `chunks`: Iterator of chunks in send order.
222    ///
223    /// # Returns
224    /// `self` for chaining.
225    pub fn stream_body<I, B>(mut self, chunks: I) -> Self
226    where
227        I: IntoIterator<Item = B>,
228        B: Into<Bytes>,
229    {
230        self.body = HttpRequestBody::Stream(chunks.into_iter().map(Into::into).collect());
231        self.streaming_body = None;
232        self
233    }
234
235    /// Sets a deferred streaming upload body factory.
236    ///
237    /// The factory runs once per send attempt and returns a fresh async byte
238    /// stream, which allows retries to rebuild the outbound stream body.
239    ///
240    /// # Parameters
241    /// - `factory`: Async stream factory for per-attempt body generation.
242    ///
243    /// # Returns
244    /// `self` for chaining.
245    pub fn streaming_body<F>(mut self, factory: F) -> Self
246    where
247        F: Fn() -> Pin<Box<dyn Future<Output = HttpRequestBodyByteStream> + Send + 'static>> + Send + Sync + 'static,
248    {
249        self.streaming_body = Some(HttpRequestStreamingBody::new(factory));
250        self.body = HttpRequestBody::Empty;
251        self
252    }
253
254    /// Sets a UTF-8 text body and adds `text/plain; charset=utf-8` if `Content-Type` is absent.
255    ///
256    /// # Parameters
257    /// - `body`: Text payload.
258    ///
259    /// # Returns
260    /// `self` for chaining.
261    pub fn text_body(mut self, body: impl Into<String>) -> Self {
262        if !self.headers.contains_key(CONTENT_TYPE) {
263            self.headers
264                .insert(CONTENT_TYPE, HeaderValue::from_static("text/plain; charset=utf-8"));
265        }
266        self.body = HttpRequestBody::Text(body.into());
267        self.streaming_body = None;
268        self
269    }
270
271    /// Serializes `value` to JSON, sets body to those bytes, and adds `application/json` if needed.
272    ///
273    /// # Parameters
274    /// - `value`: Serializable value.
275    ///
276    /// # Returns
277    /// `Ok(self)` or [`HttpError`] if JSON encoding fails.
278    pub fn json_body<T>(mut self, value: &T) -> HttpResult<Self>
279    where
280        T: Serialize,
281    {
282        let bytes = serde_json::to_vec(value)
283            .map_err(|error| HttpError::decode(format!("Failed to encode JSON body: {}", error)))?;
284        if !self.headers.contains_key(CONTENT_TYPE) {
285            self.headers
286                .insert(CONTENT_TYPE, HeaderValue::from_static("application/json"));
287        }
288        self.body = HttpRequestBody::Json(Bytes::from(bytes));
289        self.streaming_body = None;
290        Ok(self)
291    }
292
293    /// Serializes key-value pairs as `application/x-www-form-urlencoded`.
294    ///
295    /// # Parameters
296    /// - `fields`: Iterable of `(key, value)` string pairs.
297    ///
298    /// # Returns
299    /// `self` for chaining.
300    pub fn form_body<'a, I>(mut self, fields: I) -> Self
301    where
302        I: IntoIterator<Item = (&'a str, &'a str)>,
303    {
304        let mut serializer = form_urlencoded::Serializer::new(String::new());
305        for (key, value) in fields {
306            serializer.append_pair(key, value);
307        }
308        let body = serializer.finish();
309        if !self.headers.contains_key(CONTENT_TYPE) {
310            self.headers.insert(
311                CONTENT_TYPE,
312                HeaderValue::from_static("application/x-www-form-urlencoded"),
313            );
314        }
315        self.body = HttpRequestBody::Form(Bytes::from(body));
316        self.streaming_body = None;
317        self
318    }
319
320    /// Sets multipart body bytes and ensures Content-Type boundary consistency.
321    ///
322    /// # Parameters
323    /// - `body`: Multipart payload bytes.
324    /// - `boundary`: Token-safe multipart boundary used in payload framing.
325    ///
326    /// # Returns
327    /// `Ok(self)` for chaining.
328    ///
329    /// # Errors
330    /// Returns [`HttpError`] when `boundary` is not a 1 to 70 character
331    /// ASCII token-safe multipart boundary, or when an existing `Content-Type`
332    /// is not UTF-8 multipart content with a valid matching boundary.
333    pub fn multipart_body(mut self, body: impl Into<Bytes>, boundary: &str) -> HttpResult<Self> {
334        if !content_type::is_valid_multipart_boundary(boundary) {
335            return Err(HttpError::other(
336                "Invalid multipart boundary for multipart_body: expected 1 to 70 token-safe ASCII characters",
337            ));
338        }
339        if let Some(existing) = self.headers.get(CONTENT_TYPE) {
340            let existing = existing.to_str().map_err(|error| {
341                HttpError::other(format!("Existing multipart Content-Type must be valid UTF-8: {error}"))
342            })?;
343            if !content_type::is_multipart(existing) {
344                return Err(HttpError::other(
345                    "Existing Content-Type must be multipart when using multipart_body",
346                ));
347            }
348            let declares_boundary = content_type::has_parameter_name(existing, "boundary")
349                .ok_or_else(|| HttpError::other("Existing multipart Content-Type boundary is malformed or invalid"))?;
350            if let Some(existing_boundary) = content_type::parameter(existing, "boundary") {
351                if !content_type::is_valid_multipart_boundary(&existing_boundary) {
352                    return Err(HttpError::other(
353                        "Existing multipart Content-Type boundary is malformed or invalid",
354                    ));
355                }
356                if existing_boundary != boundary {
357                    return Err(HttpError::other(format!(
358                        "Existing multipart Content-Type boundary '{existing_boundary}' does not match multipart_body boundary '{boundary}'"
359                    )));
360                }
361            } else if declares_boundary {
362                return Err(HttpError::other(
363                    "Existing multipart Content-Type boundary is malformed or invalid",
364                ));
365            } else {
366                let value = existing.trim().trim_end_matches(';').trim_end();
367                let value = HeaderValue::from_str(&format!("{value}; boundary={boundary}"))
368                    .expect("validated multipart boundary should build a valid Content-Type");
369                self.headers.insert(CONTENT_TYPE, value);
370            }
371        } else {
372            let value = HeaderValue::from_str(&format!("multipart/form-data; boundary={boundary}"))
373                .expect("validated multipart boundary should build a valid Content-Type");
374            self.headers.insert(CONTENT_TYPE, value);
375        }
376        self.body = HttpRequestBody::Multipart(body.into());
377        self.streaming_body = None;
378        Ok(self)
379    }
380
381    /// Serializes records as NDJSON (`one JSON object per line`).
382    ///
383    /// # Parameters
384    /// - `records`: Serializable records to encode as NDJSON lines.
385    ///
386    /// # Returns
387    /// `Ok(self)` for chaining.
388    ///
389    /// # Errors
390    /// Returns [`HttpError`] when any record fails JSON serialization.
391    pub fn ndjson_body<T>(mut self, records: &[T]) -> HttpResult<Self>
392    where
393        T: Serialize,
394    {
395        let mut payload = String::new();
396        for record in records {
397            let line = serde_json::to_string(record)
398                .map_err(|error| HttpError::decode(format!("Failed to encode NDJSON record: {error}")))?;
399            payload.push_str(&line);
400            payload.push('\n');
401        }
402        if !self.headers.contains_key(CONTENT_TYPE) {
403            self.headers
404                .insert(CONTENT_TYPE, HeaderValue::from_static("application/x-ndjson"));
405        }
406        self.body = HttpRequestBody::Ndjson(Bytes::from(payload));
407        self.streaming_body = None;
408        Ok(self)
409    }
410
411    /// Overrides the client-wide request timeout for this request only.
412    ///
413    /// This sets reqwest's per-request [`reqwest::RequestBuilder::timeout`], i.e. a
414    /// whole-request deadline for that HTTP call (see reqwest docs for exact semantics).
415    ///
416    /// # Parameters
417    /// - `timeout`: Maximum time for the whole request (reqwest `timeout`).
418    ///
419    /// # Returns
420    /// `Ok(self)` for chaining.
421    ///
422    /// # Errors
423    /// Returns [`HttpError`] when `timeout` is zero.
424    pub fn request_timeout(mut self, timeout: Duration) -> HttpResult<Self> {
425        validate_positive_timeout("request_timeout", timeout)?;
426        self.request_timeout = Some(timeout);
427        Ok(self)
428    }
429
430    /// Overrides the write-phase timeout for this request only.
431    ///
432    /// # Parameters
433    /// - `timeout`: Maximum time allowed for sending the request bytes.
434    ///
435    /// # Returns
436    /// `Ok(self)` for chaining.
437    ///
438    /// # Errors
439    /// Returns [`HttpError`] when `timeout` is zero.
440    pub fn write_timeout(mut self, timeout: Duration) -> HttpResult<Self> {
441        validate_positive_timeout("write_timeout", timeout)?;
442        self.write_timeout = timeout;
443        Ok(self)
444    }
445
446    /// Overrides the read-phase timeout for this request only.
447    ///
448    /// # Parameters
449    /// - `timeout`: Maximum time allowed for one read wait on response body.
450    ///
451    /// # Returns
452    /// `Ok(self)` for chaining.
453    ///
454    /// # Errors
455    /// Returns [`HttpError`] when `timeout` is zero.
456    pub fn read_timeout(mut self, timeout: Duration) -> HttpResult<Self> {
457        validate_positive_timeout("read_timeout", timeout)?;
458        self.read_timeout = timeout;
459        Ok(self)
460    }
461
462    /// Overrides the client default base URL for this request.
463    ///
464    /// # Parameters
465    /// - `base_url`: Base URL used when resolving relative request paths.
466    ///
467    /// # Returns
468    /// `self` for chaining.
469    pub fn base_url(mut self, base_url: Url) -> Self {
470        self.base_url = Some(base_url);
471        self
472    }
473
474    /// Clears the base URL for this request.
475    ///
476    /// # Returns
477    /// `self` for chaining.
478    pub fn clear_base_url(mut self) -> Self {
479        self.base_url = None;
480        self
481    }
482
483    /// Overrides whether this request enforces IPv4-only literal-host validation.
484    ///
485    /// # Parameters
486    /// - `enabled`: `true` to reject IPv6 literal hosts, `false` to allow them.
487    ///
488    /// # Returns
489    /// `self` for chaining.
490    pub fn ipv4_only(mut self, enabled: bool) -> Self {
491        self.ipv4_only = enabled;
492        self
493    }
494
495    /// Binds a [`CancellationToken`] to this request.
496    ///
497    /// # Parameters
498    /// - `token`: Cancellation token checked before send and during request/stream I/O.
499    ///
500    /// # Returns
501    /// `self` for chaining.
502    pub fn cancellation_token(mut self, token: CancellationToken) -> Self {
503        self.cancellation_token = Some(token);
504        self
505    }
506
507    /// Forces retry enabled for this request even if client-level retry is disabled.
508    ///
509    /// # Returns
510    /// `self` for chaining.
511    pub fn force_retry(mut self) -> Self {
512        self.retry_override = self.retry_override.force_enable();
513        self
514    }
515
516    /// Disables retry for this request even if client-level retry is enabled.
517    ///
518    /// # Returns
519    /// `self` for chaining.
520    pub fn disable_retry(mut self) -> Self {
521        self.retry_override = self.retry_override.force_disable();
522        self
523    }
524
525    /// Overrides retryable-method policy for this request.
526    ///
527    /// # Parameters
528    /// - `policy`: Method policy to apply on this request only.
529    ///
530    /// # Returns
531    /// `self` for chaining.
532    pub fn retry_method_policy(mut self, policy: HttpRetryMethodPolicy) -> Self {
533        self.retry_override = self.retry_override.with_method_policy(policy);
534        self
535    }
536
537    /// Enables or disables honoring `Retry-After` for this request.
538    ///
539    /// # Parameters
540    /// - `enabled`: `true` to honor `Retry-After` on retryable status
541    ///   responses (`429` and `5xx`).
542    ///
543    /// # Returns
544    /// `self` for chaining.
545    pub fn honor_retry_after(mut self, enabled: bool) -> Self {
546        self.retry_override = self.retry_override.with_honor_retry_after(enabled);
547        self
548    }
549
550    /// Consumes the builder into a frozen [`HttpRequest`].
551    ///
552    /// # Returns
553    /// Built [`HttpRequest`].
554    pub fn build(self) -> HttpRequest {
555        HttpRequest::new(self)
556    }
557
558    /// Returns the URL this builder would send if it can be resolved now.
559    ///
560    /// # Returns
561    /// Resolved URL including builder query pairs, or `None` when unresolved.
562    fn debug_resolved_url(&self) -> Option<Url> {
563        let mut url = match Url::parse(&self.path) {
564            Ok(url) => url,
565            Err(_) => self.base_url.as_ref()?.join(&self.path).ok()?,
566        };
567        if !self.query.is_empty() {
568            let mut pairs = url.query_pairs_mut();
569            for (key, value) in &self.query {
570                pairs.append_pair(key, value);
571            }
572        }
573        Some(url)
574    }
575}
qubit_http/request/http_request_builder.rs

qubit_http/request/
http_request_builder.rs
