1use crate::{join_url, ApiError, ClientConfig, OAuthTokenProvider, RequestOptions};
2use base64::Engine;
3use futures::{Stream, StreamExt};
4use reqwest::{
5 header::{HeaderMap, HeaderName, HeaderValue},
6 Client, Method, Request, Response,
7};
8use serde::de::DeserializeOwned;
9use serde::de::Error as SerdeError;
10
11use std::{
12 pin::Pin,
13 str::FromStr,
14 sync::Arc,
15 task::{Context, Poll},
16};
17
18#[derive(Debug)]
21pub struct RawResponse<T> {
22 pub body: T,
24 pub status_code: u16,
26 pub headers: HeaderMap,
28}
29
30pub struct ByteStream {
32 content_length: Option<u64>,
33 inner: Pin<Box<dyn Stream<Item = Result<bytes::Bytes, reqwest::Error>> + Send>>,
34}
35
36impl ByteStream {
37 pub(crate) fn new(response: Response) -> Self {
39 let content_length = response.content_length();
40 let stream = response.bytes_stream();
41
42 Self {
43 content_length,
44 inner: Box::pin(stream),
45 }
46 }
47
48 pub async fn collect(mut self) -> Result<Vec<u8>, ApiError> {
59 let mut result = Vec::new();
60 while let Some(chunk) = self.inner.next().await {
61 result.extend_from_slice(&chunk.map_err(ApiError::Network)?);
62 }
63 Ok(result)
64 }
65
66 pub async fn try_next(&mut self) -> Result<Option<bytes::Bytes>, ApiError> {
79 match self.inner.next().await {
80 Some(Ok(bytes)) => Ok(Some(bytes)),
81 Some(Err(e)) => Err(ApiError::Network(e)),
82 None => Ok(None),
83 }
84 }
85
86 pub fn content_length(&self) -> Option<u64> {
88 self.content_length
89 }
90}
91
92impl Stream for ByteStream {
93 type Item = Result<bytes::Bytes, ApiError>;
94
95 fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
96 match self.inner.as_mut().poll_next(cx) {
97 Poll::Ready(Some(Ok(bytes))) => Poll::Ready(Some(Ok(bytes))),
98 Poll::Ready(Some(Err(e))) => Poll::Ready(Some(Err(ApiError::Network(e)))),
99 Poll::Ready(None) => Poll::Ready(None),
100 Poll::Pending => Poll::Pending,
101 }
102 }
103}
104
105#[derive(Clone)]
110pub struct OAuthConfig {
111 pub token_provider: Arc<OAuthTokenProvider>,
113 pub token_endpoint: String,
115}
116
117#[derive(Debug, Clone, serde::Deserialize)]
119struct OAuthTokenResponse {
120 access_token: String,
121 #[serde(default)]
122 expires_in: Option<i64>,
123}
124
125#[derive(Clone)]
127pub struct HttpClient {
128 client: Client,
129 config: ClientConfig,
130 oauth_config: Option<OAuthConfig>,
132}
133
134impl HttpClient {
135 pub fn new(config: ClientConfig) -> Result<Self, ApiError> {
137 Self::new_with_oauth(config, None)
138 }
139
140 pub fn new_with_oauth(
145 config: ClientConfig,
146 oauth_config: Option<OAuthConfig>,
147 ) -> Result<Self, ApiError> {
148 let client = Client::builder()
149 .timeout(config.timeout)
150 .user_agent(&config.user_agent)
151 .build()
152 .map_err(ApiError::Network)?;
153
154 Ok(Self {
155 client,
156 config,
157 oauth_config,
158 })
159 }
160
161 pub fn base_url(&self) -> &str {
163 &self.config.base_url
164 }
165
166 pub fn config(&self) -> &ClientConfig {
168 &self.config
169 }
170
171 pub async fn execute_request_raw<T>(
177 &self,
178 method: Method,
179 path: &str,
180 body: Option<serde_json::Value>,
181 query_params: Option<Vec<(String, String)>>,
182 options: Option<RequestOptions>,
183 ) -> Result<RawResponse<T>, ApiError>
184 where
185 T: DeserializeOwned,
186 {
187 let url = join_url(&self.config.base_url, path);
188 let mut request = self.client.request(method, &url);
189
190 if let Some(params) = query_params {
191 request = request.query(¶ms);
192 }
193
194 if let Some(opts) = &options {
195 if !opts.additional_query_params.is_empty() {
196 request = request.query(&opts.additional_query_params);
197 }
198 }
199
200 if let Some(body) = body {
201 request = request.json(&body);
202 }
203
204 let mut req = request.build().map_err(|e| ApiError::Network(e))?;
205
206 self.apply_auth_headers(&mut req, &options).await?;
207 self.apply_custom_headers(&mut req, &options)?;
208
209 let response = self.execute_with_retries(req, &options).await?;
210 self.parse_response_raw(response).await
211 }
212
213 pub async fn execute_request<T>(
215 &self,
216 method: Method,
217 path: &str,
218 body: Option<serde_json::Value>,
219 query_params: Option<Vec<(String, String)>>,
220 options: Option<RequestOptions>,
221 ) -> Result<T, ApiError>
222 where
223 T: DeserializeOwned, {
225 let url = join_url(&self.config.base_url, path);
226 let mut request = self.client.request(method, &url);
227
228 if let Some(params) = query_params {
230 request = request.query(¶ms);
231 }
232
233 if let Some(opts) = &options {
235 if !opts.additional_query_params.is_empty() {
236 request = request.query(&opts.additional_query_params);
237 }
238 }
239
240 if let Some(body) = body {
242 request = request.json(&body);
243 }
244
245 let mut req = request.build().map_err(|e| ApiError::Network(e))?;
247
248 self.apply_auth_headers(&mut req, &options).await?;
250 self.apply_custom_headers(&mut req, &options)?;
251
252 let response = self.execute_with_retries(req, &options).await?;
254 self.parse_response(response).await
255 }
256
257 pub async fn execute_request_with_base_url<T>(
262 &self,
263 base_url: &str,
264 method: Method,
265 path: &str,
266 body: Option<serde_json::Value>,
267 query_params: Option<Vec<(String, String)>>,
268 options: Option<RequestOptions>,
269 ) -> Result<T, ApiError>
270 where
271 T: DeserializeOwned,
272 {
273 let url = join_url(base_url, path);
274 let mut request = self.client.request(method, &url);
275
276 if let Some(params) = query_params {
277 request = request.query(¶ms);
278 }
279
280 if let Some(opts) = &options {
281 if !opts.additional_query_params.is_empty() {
282 request = request.query(&opts.additional_query_params);
283 }
284 }
285
286 if let Some(body) = body {
287 request = request.json(&body);
288 }
289
290 let mut req = request.build().map_err(|e| ApiError::Network(e))?;
291
292 self.apply_auth_headers(&mut req, &options).await?;
293 self.apply_custom_headers(&mut req, &options)?;
294
295 let response = self.execute_with_retries(req, &options).await?;
296 self.parse_response(response).await
297 }
298
299 #[cfg(feature = "multipart")]
318 pub async fn execute_multipart_request<T>(
319 &self,
320 method: Method,
321 path: &str,
322 form: reqwest::multipart::Form,
323 query_params: Option<Vec<(String, String)>>,
324 options: Option<RequestOptions>,
325 ) -> Result<T, ApiError>
326 where
327 T: DeserializeOwned,
328 {
329 let url = join_url(&self.config.base_url, path);
330 let mut request = self.client.request(method, &url);
331
332 if let Some(params) = query_params {
334 request = request.query(¶ms);
335 }
336
337 if let Some(opts) = &options {
339 if !opts.additional_query_params.is_empty() {
340 request = request.query(&opts.additional_query_params);
341 }
342 }
343
344 request = request.multipart(form);
346
347 let mut req = request.build().map_err(|e| ApiError::Network(e))?;
349
350 self.apply_auth_headers(&mut req, &options).await?;
352 self.apply_custom_headers(&mut req, &options)?;
353
354 let response = self.client.execute(req).await.map_err(ApiError::Network)?;
356
357 if !response.status().is_success() {
359 let status_code = response.status().as_u16();
360 let body = response.text().await.ok();
361 return Err(ApiError::from_response(status_code, body.as_deref()));
362 }
363
364 self.parse_response(response).await
365 }
366
367 async fn apply_auth_headers(
368 &self,
369 request: &mut Request,
370 options: &Option<RequestOptions>,
371 ) -> Result<(), ApiError> {
372 let headers = request.headers_mut();
373
374 let api_key = options
376 .as_ref()
377 .and_then(|opts| opts.api_key.as_ref())
378 .or(self.config.api_key.as_ref());
379
380 if let Some(key) = api_key {
381 let header_value = key.to_string();
382 headers.insert(
383 "api_key",
384 header_value.parse().map_err(|_| ApiError::InvalidHeader)?,
385 );
386 }
387
388 let token = if let Some(opts) = options.as_ref() {
390 if opts.token.is_some() {
391 opts.token.clone()
392 } else {
393 None
394 }
395 } else {
396 None
397 };
398
399 let token = match token {
400 Some(t) => Some(t),
401 None => {
402 if let Some(oauth_config) = &self.oauth_config {
404 Some(self.get_oauth_token(oauth_config).await?)
405 } else {
406 self.config.token.clone()
408 }
409 }
410 };
411
412 if let Some(token) = token {
413 let auth_value = format!("Bearer {}", token);
414 headers.insert(
415 "Authorization",
416 auth_value.parse().map_err(|_| ApiError::InvalidHeader)?,
417 );
418 }
419
420 Ok(())
421 }
422
423 async fn get_oauth_token(&self, oauth_config: &OAuthConfig) -> Result<String, ApiError> {
425 let token_provider = &oauth_config.token_provider;
426 let token_endpoint = &oauth_config.token_endpoint;
427 let client_id = token_provider.client_id().to_string();
428 let client_secret = token_provider.client_secret().to_string();
429 let base_url = self.config.base_url.clone();
430
431 token_provider
433 .get_or_fetch_async(|| async {
434 self.fetch_oauth_token(&base_url, token_endpoint, &client_id, &client_secret)
435 .await
436 })
437 .await
438 }
439
440 async fn fetch_oauth_token(
442 &self,
443 base_url: &str,
444 token_endpoint: &str,
445 client_id: &str,
446 client_secret: &str,
447 ) -> Result<(String, u64), ApiError> {
448 let url = join_url(base_url, token_endpoint);
449
450 let body = serde_json::json!({
452 "client_id": client_id,
453 "client_secret": client_secret,
454 "grant_type": "client_credentials"
455 });
456
457 let response = self
458 .client
459 .request(Method::POST, &url)
460 .json(&body)
461 .send()
462 .await
463 .map_err(ApiError::Network)?;
464
465 if !response.status().is_success() {
466 let status_code = response.status().as_u16();
467 let body = response.text().await.ok();
468 return Err(ApiError::from_response(status_code, body.as_deref()));
469 }
470
471 let token_response: OAuthTokenResponse =
473 response.json().await.map_err(ApiError::Network)?;
474
475 let expires_in = token_response.expires_in.unwrap_or(3600) as u64;
476 Ok((token_response.access_token, expires_in))
477 }
478
479 fn apply_custom_headers(
480 &self,
481 request: &mut Request,
482 options: &Option<RequestOptions>,
483 ) -> Result<(), ApiError> {
484 let headers = request.headers_mut();
485
486 for (key, value) in &self.config.custom_headers {
488 headers.insert(
489 HeaderName::from_str(key).map_err(|_| ApiError::InvalidHeader)?,
490 HeaderValue::from_str(value).map_err(|_| ApiError::InvalidHeader)?,
491 );
492 }
493
494 if let Some(options) = options {
496 for (key, value) in &options.additional_headers {
497 headers.insert(
498 HeaderName::from_str(key).map_err(|_| ApiError::InvalidHeader)?,
499 HeaderValue::from_str(value).map_err(|_| ApiError::InvalidHeader)?,
500 );
501 }
502 }
503
504 Ok(())
505 }
506
507 async fn execute_with_retries(
508 &self,
509 request: Request,
510 options: &Option<RequestOptions>,
511 ) -> Result<Response, ApiError> {
512 let max_retries = options
513 .as_ref()
514 .and_then(|opts| opts.max_retries)
515 .unwrap_or(self.config.max_retries);
516
517 let mut last_error = None;
518
519 for attempt in 0..=max_retries {
520 let cloned_request = request.try_clone().ok_or(ApiError::RequestClone)?;
521
522 match self.client.execute(cloned_request).await {
523 Ok(response) if response.status().is_success() => return Ok(response),
524 Ok(response)
525 if attempt < max_retries
526 && Self::is_retryable_status(response.status().as_u16()) =>
527 {
528 let delay = std::time::Duration::from_millis(100 * 2_u64.pow(attempt));
530 tokio::time::sleep(delay).await;
531 }
532 Ok(response) => {
533 let status_code = response.status().as_u16();
534 let body = response.text().await.ok();
535 return Err(ApiError::from_response(status_code, body.as_deref()));
536 }
537 Err(e) if attempt < max_retries => {
538 last_error = Some(e);
539 let delay = std::time::Duration::from_millis(100 * 2_u64.pow(attempt));
541 tokio::time::sleep(delay).await;
542 }
543 Err(e) => return Err(ApiError::Network(e)),
544 }
545 }
546
547 Err(ApiError::Network(last_error.unwrap()))
548 }
549
550 fn is_retryable_status(status_code: u16) -> bool {
551 [408, 429].contains(&status_code) || status_code >= 500
552 }
553
554 async fn parse_response<T>(&self, response: Response) -> Result<T, ApiError>
555 where
556 T: DeserializeOwned,
557 {
558 let status = response.status().as_u16();
559 let text = response.text().await.map_err(ApiError::Network)?;
560
561 if text.is_empty() {
563 return Err(ApiError::Http {
564 status,
565 message: String::new(),
566 });
567 }
568
569 serde_json::from_str(&text).map_err(ApiError::Serialization)
570 }
571
572 async fn parse_response_raw<T>(&self, response: Response) -> Result<RawResponse<T>, ApiError>
573 where
574 T: DeserializeOwned,
575 {
576 let status_code = response.status().as_u16();
577 let headers = response.headers().clone();
578 let text = response.text().await.map_err(ApiError::Network)?;
579
580 if text.is_empty() {
581 return Err(ApiError::Http {
582 status: status_code,
583 message: String::new(),
584 });
585 }
586
587 let body: T = serde_json::from_str(&text).map_err(ApiError::Serialization)?;
588 Ok(RawResponse {
589 body,
590 status_code,
591 headers,
592 })
593 }
594
595 pub async fn execute_request_base64(
601 &self,
602 method: Method,
603 path: &str,
604 body: Option<serde_json::Value>,
605 query_params: Option<Vec<(String, String)>>,
606 options: Option<RequestOptions>,
607 ) -> Result<Vec<u8>, ApiError> {
608 let url = join_url(&self.config.base_url, path);
609 let mut request = self.client.request(method, &url);
610
611 if let Some(params) = query_params {
613 request = request.query(¶ms);
614 }
615
616 if let Some(opts) = &options {
618 if !opts.additional_query_params.is_empty() {
619 request = request.query(&opts.additional_query_params);
620 }
621 }
622
623 if let Some(body) = body {
625 request = request.json(&body);
626 }
627
628 let mut req = request.build().map_err(|e| ApiError::Network(e))?;
630
631 self.apply_auth_headers(&mut req, &options).await?;
633 self.apply_custom_headers(&mut req, &options)?;
634
635 let response = self.execute_with_retries(req, &options).await?;
637
638 let text = response.text().await.map_err(ApiError::Network)?;
640 let base64_string: String = serde_json::from_str(&text).map_err(ApiError::Serialization)?;
641 base64::engine::general_purpose::STANDARD
642 .decode(&base64_string)
643 .map_err(|e| {
644 ApiError::Serialization(SerdeError::custom(format!("base64 decode error: {}", e)))
645 })
646 }
647
648 pub async fn execute_stream_request(
704 &self,
705 method: Method,
706 path: &str,
707 body: Option<serde_json::Value>,
708 query_params: Option<Vec<(String, String)>>,
709 options: Option<RequestOptions>,
710 ) -> Result<ByteStream, ApiError> {
711 let url = join_url(&self.config.base_url, path);
712 let mut request = self.client.request(method, &url);
713
714 if let Some(params) = query_params {
716 request = request.query(¶ms);
717 }
718
719 if let Some(opts) = &options {
721 if !opts.additional_query_params.is_empty() {
722 request = request.query(&opts.additional_query_params);
723 }
724 }
725
726 if let Some(body) = body {
728 request = request.json(&body);
729 }
730
731 let mut req = request.build().map_err(|e| ApiError::Network(e))?;
733
734 self.apply_auth_headers(&mut req, &options).await?;
736 self.apply_custom_headers(&mut req, &options)?;
737
738 let response = self.execute_with_retries(req, &options).await?;
740
741 Ok(ByteStream::new(response))
743 }
744
745 pub async fn execute_stream_request_with_base_url(
747 &self,
748 base_url: &str,
749 method: Method,
750 path: &str,
751 body: Option<serde_json::Value>,
752 query_params: Option<Vec<(String, String)>>,
753 options: Option<RequestOptions>,
754 ) -> Result<ByteStream, ApiError> {
755 let url = join_url(base_url, path);
756 let mut request = self.client.request(method, &url);
757
758 if let Some(params) = query_params {
759 request = request.query(¶ms);
760 }
761
762 if let Some(opts) = &options {
763 if !opts.additional_query_params.is_empty() {
764 request = request.query(&opts.additional_query_params);
765 }
766 }
767
768 if let Some(body) = body {
769 request = request.json(&body);
770 }
771
772 let mut req = request.build().map_err(|e| ApiError::Network(e))?;
773
774 self.apply_auth_headers(&mut req, &options).await?;
775 self.apply_custom_headers(&mut req, &options)?;
776
777 let response = self.execute_with_retries(req, &options).await?;
778
779 Ok(ByteStream::new(response))
780 }
781}
782
783#[cfg(test)]
784mod tests {
785 use super::*;
786
787 #[test]
788 fn test_is_retryable_status() {
789 assert!(HttpClient::is_retryable_status(408));
791 assert!(HttpClient::is_retryable_status(429));
792
793 assert!(HttpClient::is_retryable_status(500));
795 assert!(HttpClient::is_retryable_status(501));
796 assert!(HttpClient::is_retryable_status(502));
797 assert!(HttpClient::is_retryable_status(503));
798 assert!(HttpClient::is_retryable_status(504));
799 assert!(HttpClient::is_retryable_status(599));
800
801 assert!(!HttpClient::is_retryable_status(200));
803 assert!(!HttpClient::is_retryable_status(400));
804 assert!(!HttpClient::is_retryable_status(401));
805 assert!(!HttpClient::is_retryable_status(404));
806 }
807}