rig_core/http_client/
mod.rs

1use crate::http_client::sse::BoxedStream;
2use bytes::Bytes;
3pub use http::{HeaderMap, HeaderValue, Method, Request, Response, Uri, request::Builder};
4use http::{HeaderName, StatusCode};
5use reqwest::Body;
6pub mod multipart;
7pub mod retry;
8pub mod sse;
9use crate::wasm_compat::*;
10pub use multipart::MultipartForm;
11pub use reqwest::Client as ReqwestClient;
12use std::pin::Pin;
13
14#[derive(Debug, thiserror::Error)]
15pub enum Error {
16    #[error("Http error: {0}")]
17    Protocol(#[from] http::Error),
18    #[error("Invalid status code: {0}")]
19    InvalidStatusCode(StatusCode),
20    #[error("Invalid status code {0} with message: {1}")]
21    InvalidStatusCodeWithMessage(StatusCode, String),
22    #[error("Header value outside of legal range: {0}")]
23    InvalidHeaderValue(#[from] http::header::InvalidHeaderValue),
24    #[error("Request in error state, cannot access headers")]
25    NoHeaders,
26    #[error("Stream ended")]
27    StreamEnded,
28    #[error("Invalid content type was returned: {0:?}")]
29    InvalidContentType(HeaderValue),
30    #[cfg(not(target_family = "wasm"))]
31    #[error("Http client error: {0}")]
32    Instance(#[from] Box<dyn std::error::Error + Send + Sync + 'static>),
33
34    #[cfg(target_family = "wasm")]
35    #[error("Http client error: {0}")]
36    Instance(#[from] Box<dyn std::error::Error + 'static>),
37}
38
39impl Error {
40    pub(crate) fn non_success_status(&self) -> Option<StatusCode> {
41        match self {
42            Self::InvalidStatusCode(status) | Self::InvalidStatusCodeWithMessage(status, _) => {
43                Some(*status)
44            }
45            _ => None,
46        }
47    }
48
49    pub(crate) fn non_success_body(&self) -> Option<&str> {
50        match self {
51            Self::InvalidStatusCodeWithMessage(_, body) => Some(body.as_str()),
52            _ => None,
53        }
54    }
55}
56
57pub type Result<T> = std::result::Result<T, Error>;
58
59#[cfg(not(target_family = "wasm"))]
60pub(crate) fn instance_error<E: std::error::Error + Send + Sync + 'static>(error: E) -> Error {
61    Error::Instance(error.into())
62}
63
64#[cfg(target_family = "wasm")]
65fn instance_error<E: std::error::Error + 'static>(error: E) -> Error {
66    Error::Instance(error.into())
67}
68
69async fn non_success_status_error(response: reqwest::Response) -> Error {
70    let status = response.status();
71    let message = response
72        .text()
73        .await
74        .unwrap_or_else(|error| format!("failed to read error response body: {error}"));
75    Error::InvalidStatusCodeWithMessage(status, message)
76}
77
78pub type LazyBytes = WasmBoxedFuture<'static, Result<Bytes>>;
79pub type LazyBody<T> = WasmBoxedFuture<'static, Result<T>>;
80
81pub type StreamingResponse = Response<BoxedStream>;
82
83#[derive(Debug, Clone, Copy)]
84pub struct NoBody;
85
86impl From<NoBody> for Bytes {
87    fn from(_: NoBody) -> Self {
88        Bytes::new()
89    }
90}
91
92impl From<NoBody> for Body {
93    fn from(_: NoBody) -> Self {
94        reqwest::Body::default()
95    }
96}
97
98pub async fn text(response: Response<LazyBody<Vec<u8>>>) -> Result<String> {
99    let text = response.into_body().await?;
100    Ok(String::from(String::from_utf8_lossy(&text)))
101}
102
103pub fn make_auth_header(key: impl AsRef<str>) -> Result<(HeaderName, HeaderValue)> {
104    Ok((
105        http::header::AUTHORIZATION,
106        HeaderValue::from_str(&format!("Bearer {}", key.as_ref()))?,
107    ))
108}
109
110pub fn bearer_auth_header(headers: &mut HeaderMap, key: impl AsRef<str>) -> Result<()> {
111    let (k, v) = make_auth_header(key)?;
112
113    headers.insert(k, v);
114
115    Ok(())
116}
117
118pub fn with_bearer_auth(mut req: Builder, auth: &str) -> Result<Builder> {
119    bearer_auth_header(req.headers_mut().ok_or(Error::NoHeaders)?, auth)?;
120
121    Ok(req)
122}
123
124/// A helper trait to make generic requests (both regular and SSE) possible.
125pub trait HttpClientExt: WasmCompatSend + WasmCompatSync {
126    /// Send a HTTP request, get a response back (as bytes). Response must be able to be turned back into Bytes.
127    fn send<T, U>(
128        &self,
129        req: Request<T>,
130    ) -> impl Future<Output = Result<Response<LazyBody<U>>>> + WasmCompatSend + 'static
131    where
132        T: Into<Bytes>,
133        T: WasmCompatSend,
134        U: From<Bytes>,
135        U: WasmCompatSend + 'static;
136
137    /// Send a HTTP request with a multipart body, get a response back (as bytes). Response must be able to be turned back into Bytes (although usually for the response, you will probably want to specify Bytes anyway).
138    fn send_multipart<U>(
139        &self,
140        req: Request<MultipartForm>,
141    ) -> impl Future<Output = Result<Response<LazyBody<U>>>> + WasmCompatSend + 'static
142    where
143        U: From<Bytes>,
144        U: WasmCompatSend + 'static;
145
146    /// Send a HTTP request, get a streamed response back (as a stream of [`bytes::Bytes`].)
147    fn send_streaming<T>(
148        &self,
149        req: Request<T>,
150    ) -> impl Future<Output = Result<StreamingResponse>> + WasmCompatSend
151    where
152        T: Into<Bytes> + WasmCompatSend;
153}
154
155impl HttpClientExt for reqwest::Client {
156    fn send<T, U>(
157        &self,
158        req: Request<T>,
159    ) -> impl Future<Output = Result<Response<LazyBody<U>>>> + WasmCompatSend + 'static
160    where
161        T: Into<Bytes>,
162        U: From<Bytes> + WasmCompatSend,
163    {
164        let (parts, body) = req.into_parts();
165        let req = self
166            .request(parts.method, parts.uri.to_string())
167            .headers(parts.headers)
168            .body(body.into());
169
170        async move {
171            let response = req.send().await.map_err(instance_error)?;
172            if !response.status().is_success() {
173                return Err(non_success_status_error(response).await);
174            }
175
176            let mut res = Response::builder().status(response.status());
177
178            if let Some(hs) = res.headers_mut() {
179                *hs = response.headers().clone();
180            }
181
182            let body: LazyBody<U> = Box::pin(async {
183                let bytes = response
184                    .bytes()
185                    .await
186                    .map_err(|e| Error::Instance(e.into()))?;
187
188                let body = U::from(bytes);
189                Ok(body)
190            });
191
192            res.body(body).map_err(Error::Protocol)
193        }
194    }
195
196    fn send_multipart<U>(
197        &self,
198        req: Request<MultipartForm>,
199    ) -> impl Future<Output = Result<Response<LazyBody<U>>>> + WasmCompatSend + 'static
200    where
201        U: From<Bytes>,
202        U: WasmCompatSend + 'static,
203    {
204        let (parts, body) = req.into_parts();
205        let body = reqwest::multipart::Form::from(body);
206
207        let req = self
208            .request(parts.method, parts.uri.to_string())
209            .headers(parts.headers)
210            .multipart(body);
211
212        async move {
213            let response = req.send().await.map_err(instance_error)?;
214            if !response.status().is_success() {
215                return Err(non_success_status_error(response).await);
216            }
217
218            let mut res = Response::builder().status(response.status());
219
220            if let Some(hs) = res.headers_mut() {
221                *hs = response.headers().clone();
222            }
223
224            let body: LazyBody<U> = Box::pin(async {
225                let bytes = response
226                    .bytes()
227                    .await
228                    .map_err(|e| Error::Instance(e.into()))?;
229
230                let body = U::from(bytes);
231                Ok(body)
232            });
233
234            res.body(body).map_err(Error::Protocol)
235        }
236    }
237
238    fn send_streaming<T>(
239        &self,
240        req: Request<T>,
241    ) -> impl Future<Output = Result<StreamingResponse>> + WasmCompatSend
242    where
243        T: Into<Bytes> + WasmCompatSend,
244    {
245        let (parts, body) = req.into_parts();
246
247        let client = self.clone();
248
249        async move {
250            let req = self
251                .request(parts.method, parts.uri.to_string())
252                .headers(parts.headers)
253                .body(body.into())
254                .build()
255                .map_err(|error| Error::Instance(error.into()))?;
256            let response: reqwest::Response = client.execute(req).await.map_err(instance_error)?;
257            if !response.status().is_success() {
258                return Err(non_success_status_error(response).await);
259            }
260
261            #[cfg(not(target_family = "wasm"))]
262            let mut res = Response::builder()
263                .status(response.status())
264                .version(response.version());
265
266            #[cfg(target_family = "wasm")]
267            let mut res = Response::builder().status(response.status());
268
269            if let Some(hs) = res.headers_mut() {
270                *hs = response.headers().clone();
271            }
272
273            use futures::StreamExt;
274
275            let mapped_stream: Pin<Box<dyn WasmCompatSendStream<InnerItem = Result<Bytes>>>> =
276                Box::pin(
277                    response
278                        .bytes_stream()
279                        .map(|chunk| chunk.map_err(|e| Error::Instance(Box::new(e)))),
280                );
281
282            res.body(mapped_stream).map_err(Error::Protocol)
283        }
284    }
285}
286
287#[cfg(feature = "reqwest-middleware")]
288#[cfg_attr(docsrs, doc(cfg(feature = "reqwest-middleware")))]
289impl HttpClientExt for reqwest_middleware::ClientWithMiddleware {
290    fn send<T, U>(
291        &self,
292        req: Request<T>,
293    ) -> impl Future<Output = Result<Response<LazyBody<U>>>> + WasmCompatSend + 'static
294    where
295        T: Into<Bytes>,
296        U: From<Bytes> + WasmCompatSend,
297    {
298        let (parts, body) = req.into_parts();
299        let req = self
300            .request(parts.method, parts.uri.to_string())
301            .headers(parts.headers)
302            .body(body.into());
303
304        async move {
305            let response = req.send().await.map_err(instance_error)?;
306            if !response.status().is_success() {
307                return Err(non_success_status_error(response).await);
308            }
309
310            let mut res = Response::builder().status(response.status());
311
312            if let Some(hs) = res.headers_mut() {
313                *hs = response.headers().clone();
314            }
315
316            let body: LazyBody<U> = Box::pin(async {
317                let bytes = response
318                    .bytes()
319                    .await
320                    .map_err(|e| Error::Instance(e.into()))?;
321
322                let body = U::from(bytes);
323                Ok(body)
324            });
325
326            res.body(body).map_err(Error::Protocol)
327        }
328    }
329
330    fn send_multipart<U>(
331        &self,
332        req: Request<MultipartForm>,
333    ) -> impl Future<Output = Result<Response<LazyBody<U>>>> + WasmCompatSend + 'static
334    where
335        U: From<Bytes>,
336        U: WasmCompatSend + 'static,
337    {
338        let (parts, body) = req.into_parts();
339        let body = reqwest::multipart::Form::from(body);
340
341        let req = self
342            .request(parts.method, parts.uri.to_string())
343            .headers(parts.headers)
344            .multipart(body);
345
346        async move {
347            let response = req.send().await.map_err(instance_error)?;
348            if !response.status().is_success() {
349                return Err(non_success_status_error(response).await);
350            }
351
352            let mut res = Response::builder().status(response.status());
353
354            if let Some(hs) = res.headers_mut() {
355                *hs = response.headers().clone();
356            }
357
358            let body: LazyBody<U> = Box::pin(async {
359                let bytes = response
360                    .bytes()
361                    .await
362                    .map_err(|e| Error::Instance(e.into()))?;
363
364                let body = U::from(bytes);
365                Ok(body)
366            });
367
368            res.body(body).map_err(Error::Protocol)
369        }
370    }
371
372    fn send_streaming<T>(
373        &self,
374        req: Request<T>,
375    ) -> impl Future<Output = Result<StreamingResponse>> + WasmCompatSend
376    where
377        T: Into<Bytes> + WasmCompatSend,
378    {
379        let (parts, body) = req.into_parts();
380
381        let client = self.clone();
382
383        async move {
384            let req = self
385                .request(parts.method, parts.uri.to_string())
386                .headers(parts.headers)
387                .body(body.into())
388                .build()
389                .map_err(|error| Error::Instance(error.into()))?;
390            let response: reqwest::Response = client.execute(req).await.map_err(instance_error)?;
391            if !response.status().is_success() {
392                return Err(non_success_status_error(response).await);
393            }
394
395            #[cfg(not(target_family = "wasm"))]
396            let mut res = Response::builder()
397                .status(response.status())
398                .version(response.version());
399
400            #[cfg(target_family = "wasm")]
401            let mut res = Response::builder().status(response.status());
402
403            if let Some(hs) = res.headers_mut() {
404                *hs = response.headers().clone();
405            }
406
407            use futures::StreamExt;
408
409            let mapped_stream: Pin<Box<dyn WasmCompatSendStream<InnerItem = Result<Bytes>>>> =
410                Box::pin(
411                    response
412                        .bytes_stream()
413                        .map(|chunk| chunk.map_err(|e| Error::Instance(Box::new(e)))),
414                );
415
416            res.body(mapped_stream).map_err(Error::Protocol)
417        }
418    }
419}
rig_core/http_client/mod.rs

rig_core/http_client/
mod.rs
