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

qubit_http/request/
http_request_builder.rs
