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

qubit_http/request/
http_request_builder.rs
