apify_rs/
client.rs

1use crate::error::ApifyError;
2use crate::models::{DataResponse, ListResponse};
3use reqwest::{Client, Method, RequestBuilder};
4use serde::de::DeserializeOwned;
5use serde::Serialize;
6use std::time::Duration;
7
8const DEFAULT_BASE_URL: &str = "https://api.apify.com/v2";
9const DEFAULT_TIMEOUT_SECS: u64 = 60;
10
11/// Low-level HTTP transport for the Apify API.
12///
13/// You usually do not interact with this directly — [`ApifyClient`](crate::ApifyClient)
14/// owns an `HttpClient` and passes references to the resource clients.
15///
16/// Responsibilities:
17/// * Base-URL + path concatenation.
18/// * Bearer-token authentication.
19/// * JSON body serialization / deserialization.
20/// * Translating HTTP errors into [`ApifyError`].
21/// * Exponential-backoff retries for **429 Too Many Requests**.
22#[derive(Debug, Clone)]
23pub struct HttpClient {
24    pub(crate) client: Client,
25    pub(crate) base_url: String,
26    pub(crate) token: Option<String>,
27}
28
29impl HttpClient {
30    /// Construct a new HTTP client.
31    ///
32    /// `token` is optional — pass `None` for anonymous (read-only) access.
33    pub fn new(token: Option<String>) -> Self {
34        let client = Client::builder()
35            .timeout(Duration::from_secs(DEFAULT_TIMEOUT_SECS))
36            .build()
37            .expect("Failed to build HTTP client");
38
39        Self {
40            client,
41            base_url: DEFAULT_BASE_URL.to_string(),
42            token,
43        }
44    }
45
46    /// Override the base URL used for every request.
47    pub fn with_base_url(mut self, base_url: impl Into<String>) -> Self {
48        self.base_url = base_url.into();
49        self
50    }
51
52    /// Re-build the underlying [`reqwest::Client`] with a custom timeout.
53    pub fn with_timeout(mut self, timeout: Duration) -> Self {
54        self.client = Client::builder()
55            .timeout(timeout)
56            .build()
57            .expect("Failed to build HTTP client");
58        self
59    }
60
61    fn build_url(&self, path: &str) -> String {
62        format!("{}/{}", self.base_url.trim_end_matches('/'), path.trim_start_matches('/'))
63    }
64
65    fn authenticate(&self, builder: RequestBuilder) -> RequestBuilder {
66        match &self.token {
67            Some(token) => builder.bearer_auth(token),
68            None => builder,
69        }
70    }
71
72    /// Send a request and deserialize the JSON body into `T`.
73    ///
74    /// `path` is relative to the base URL (e.g. `/actor-tasks`).
75    /// Authentication and JSON `Content-Type` are handled automatically.
76    pub async fn request<T: DeserializeOwned>(
77        &self,
78        method: Method,
79        path: &str,
80        body: Option<impl Serialize>,
81    ) -> Result<T, ApifyError> {
82        let url = self.build_url(path);
83        let mut builder = self.client.request(method.clone(), &url);
84        builder = self.authenticate(builder);
85
86        if let Some(body_data) = body {
87            builder = builder.json(&body_data);
88        }
89
90        let response = builder.send().await?;
91        self.handle_response(response).await
92    }
93
94    async fn handle_response<T: DeserializeOwned>(
95        &self,
96        response: reqwest::Response,
97    ) -> Result<T, ApifyError> {
98        let status = response.status();
99
100        if status.is_success() {
101            let data = response.json::<T>().await?;
102            return Ok(data);
103        }
104
105        // Attempt to parse Apify's standard error envelope:
106        // { "error": { "type": "...", "message": "..." } }
107        let error_body: serde_json::Value = response.json().await.unwrap_or_default();
108        let error_type = error_body["error"]["type"]
109            .as_str()
110            .unwrap_or("unknown")
111            .to_string();
112        let message = error_body["error"]["message"]
113            .as_str()
114            .unwrap_or("Unknown error")
115            .to_string();
116
117        Err(ApifyError::from_api_response(
118            status.as_u16(),
119            error_type,
120            message,
121        ))
122    }
123
124    /// Send a request with exponential-backoff retries on **429** responses.
125    ///
126    /// Starts with a 500 ms delay and doubles it up to 5 attempts.
127    /// Use this for idempotent reads when you expect bursts of traffic.
128    pub async fn request_with_retry<T: DeserializeOwned>(
129        &self,
130        method: Method,
131        path: &str,
132        body: Option<impl Serialize>,
133    ) -> Result<T, ApifyError> {
134        let body_value = match body {
135            Some(b) => Some(serde_json::to_value(b)?),
136            None => None,
137        };
138
139        let mut delay_ms: u64 = 500;
140        let max_retries = 5;
141
142        for attempt in 0..max_retries {
143            match self.request(method.clone(), path, body_value.clone()).await {
144                Ok(data) => return Ok(data),
145                Err(ApifyError::RateLimit) if attempt < max_retries - 1 => {
146                    tokio::time::sleep(Duration::from_millis(delay_ms)).await;
147                    delay_ms *= 2;
148                }
149                Err(e) => return Err(e),
150            }
151        }
152
153        Err(ApifyError::RateLimit)
154    }
155}
156
157/// Convenience helpers that unwrap Apify's `{ "data": ... }` envelope.
158impl HttpClient {
159    /// `GET` a single resource and return the inner `data` field.
160    pub async fn get_data<T: DeserializeOwned>(&self, path: &str) -> Result<T, ApifyError> {
161        let resp: DataResponse<T> = self.request(Method::GET, path, None::<()>).await?;
162        Ok(resp.data)
163    }
164
165    /// `POST` a JSON body and return the inner `data` field of the response.
166    pub async fn post_data<T: DeserializeOwned, B: Serialize>(
167        &self,
168        path: &str,
169        body: B,
170    ) -> Result<T, ApifyError> {
171        let resp: DataResponse<T> = self.request(Method::POST, path, Some(body)).await?;
172        Ok(resp.data)
173    }
174
175    /// `PUT` a JSON body and return the inner `data` field of the response.
176    pub async fn put_data<T: DeserializeOwned, B: Serialize>(
177        &self,
178        path: &str,
179        body: B,
180    ) -> Result<T, ApifyError> {
181        let resp: DataResponse<T> = self.request(Method::PUT, path, Some(body)).await?;
182        Ok(resp.data)
183    }
184
185    /// `DELETE` a resource, returning `()` on success.
186    pub async fn delete_request(&self, path: &str) -> Result<(), ApifyError> {
187        let resp = self
188            .client
189            .delete(self.build_url(path))
190            .send()
191            .await?;
192
193        if resp.status().is_success() {
194            Ok(())
195        } else {
196            let status = resp.status();
197            let error_body: serde_json::Value = resp.json().await.unwrap_or_default();
198            let error_type = error_body["error"]["type"]
199                .as_str()
200                .unwrap_or("unknown")
201                .to_string();
202            let message = error_body["error"]["message"]
203                .as_str()
204                .unwrap_or("Unknown error")
205                .to_string();
206            Err(ApifyError::from_api_response(status.as_u16(), error_type, message))
207        }
208    }
209
210    /// `GET` a paginated list and return the inner [`ListData`](crate::models::ListData).
211    pub async fn get_list<T: DeserializeOwned>(
212        &self,
213        path: &str,
214    ) -> Result<crate::models::ListData<T>, ApifyError> {
215        let resp: ListResponse<T> = self.request(Method::GET, path, None::<()>).await?;
216        Ok(resp.data)
217    }
218}
219
220#[cfg(test)]
221mod tests {
222    use super::*;
223    use crate::models::{ActorJobStatus, DataResponse, ListData, ListResponse, Run, Task, TaskShort};
224    use chrono::Utc;
225
226    #[tokio::test]
227    async fn test_get_task_success() {
228        let mut server = mockito::Server::new_async().await;
229
230        let task = Task {
231            id: "test-task-id".to_string(),
232            user_id: "user123".to_string(),
233            act_id: "actor456".to_string(),
234            name: "test-task".to_string(),
235            username: Some("testuser".to_string()),
236            created_at: Utc::now(),
237            modified_at: Utc::now(),
238            removed_at: None,
239            stats: None,
240            options: None,
241            input: None,
242            title: None,
243        };
244
245        let response = DataResponse { data: task };
246        let body = serde_json::to_string(&response).unwrap();
247
248        let mock = server
249            .mock("GET", "/actor-tasks/test-task-id")
250            .with_status(200)
251            .with_header("content-type", "application/json")
252            .with_body(body)
253            .create_async()
254            .await;
255
256        let client = HttpClient::new(Some("test-token".to_string()))
257            .with_base_url(server.url());
258
259        let result = client.get_data::<Task>("/actor-tasks/test-task-id").await;
260        assert!(result.is_ok());
261        let task = result.unwrap();
262        assert_eq!(task.id, "test-task-id");
263        assert_eq!(task.name, "test-task");
264
265        mock.assert_async().await;
266    }
267
268    #[tokio::test]
269    async fn test_get_run_success() {
270        let mut server = mockito::Server::new_async().await;
271
272        let run = Run {
273            id: "run-123".to_string(),
274            act_id: "actor-456".to_string(),
275            user_id: "user-789".to_string(),
276            actor_task_id: Some("task-abc".to_string()),
277            started_at: Utc::now(),
278            finished_at: None,
279            status: ActorJobStatus::Running,
280            status_message: Some("Actor is running".to_string()),
281            is_status_message_terminal: Some(false),
282            meta: None,
283            stats: None,
284            options: None,
285            build_id: "build-xyz".to_string(),
286            exit_code: None,
287            default_key_value_store_id: "kv-store-1".to_string(),
288            default_dataset_id: "dataset-1".to_string(),
289            default_request_queue_id: "queue-1".to_string(),
290            storage_ids: None,
291            build_number: Some("0.1.0".to_string()),
292            container_url: None,
293            is_container_server_ready: None,
294            git_branch_name: None,
295            usage_total_usd: None,
296            charged_event_counts: None,
297        };
298
299        let response = DataResponse { data: run };
300        let body = serde_json::to_string(&response).unwrap();
301
302        let mock = server
303            .mock("GET", "/actor-runs/run-123")
304            .with_status(200)
305            .with_header("content-type", "application/json")
306            .with_body(body)
307            .create_async()
308            .await;
309
310        let client = HttpClient::new(Some("test-token".to_string()))
311            .with_base_url(server.url());
312
313        let result = client.get_data::<Run>("/actor-runs/run-123").await;
314        assert!(result.is_ok());
315        let run = result.unwrap();
316        assert_eq!(run.id, "run-123");
317        assert_eq!(run.status, ActorJobStatus::Running);
318
319        mock.assert_async().await;
320    }
321
322    #[tokio::test]
323    async fn test_api_error_response() {
324        let mut server = mockito::Server::new_async().await;
325
326        let error_body = serde_json::json!({
327            "error": {
328                "type": "record-not-found",
329                "message": "Store was not found."
330            }
331        });
332
333        let mock = server
334            .mock("GET", "/actor-runs/nonexistent")
335            .with_status(404)
336            .with_header("content-type", "application/json")
337            .with_body(error_body.to_string())
338            .create_async()
339            .await;
340
341        let client = HttpClient::new(Some("test-token".to_string()))
342            .with_base_url(server.url());
343
344        let result = client.get_data::<Run>("/actor-runs/nonexistent").await;
345        assert!(result.is_err());
346        let err = result.unwrap_err();
347        let err_str = err.to_string();
348        assert!(err_str.contains("not found") || err_str.contains("Store was not found"));
349
350        mock.assert_async().await;
351    }
352
353    #[tokio::test]
354    async fn test_list_tasks() {
355        let mut server = mockito::Server::new_async().await;
356
357        let tasks = vec![TaskShort {
358            id: "task-1".to_string(),
359            act_id: "actor-1".to_string(),
360            name: "Task One".to_string(),
361            username: None,
362            created_at: Utc::now(),
363            modified_at: Utc::now(),
364        }];
365
366        let response = ListResponse {
367            data: ListData {
368                total: 1,
369                offset: 0,
370                limit: 10,
371                count: 1,
372                desc: false,
373                items: tasks,
374            },
375        };
376
377        let mock = server
378            .mock("GET", "/actor-tasks")
379            .with_status(200)
380            .with_header("content-type", "application/json")
381            .with_body(serde_json::to_string(&response).unwrap())
382            .create_async()
383            .await;
384
385        let client = HttpClient::new(Some("test-token".to_string()))
386            .with_base_url(server.url());
387
388        let result = client.get_list::<TaskShort>("/actor-tasks").await;
389        assert!(result.is_ok());
390        let list = result.unwrap();
391        assert_eq!(list.total, 1);
392        assert_eq!(list.items.len(), 1);
393        assert_eq!(list.items[0].name, "Task One");
394
395        mock.assert_async().await;
396    }
397}
apify_rs/client.rs

apify_rs/
client.rs
