apify_client/clients/
request_queue.rs

1//! Client for a single request queue (`/v2/request-queues/{queueId}` and variants).
2
3use serde::Serialize;
4
5use crate::clients::base::{
6    delete_resource, delete_with_body, get_resource, get_resource_required, post_action,
7    post_with_body, update_resource, ResourceContext,
8};
9use crate::common::{encode_path_segment, QueryParams};
10use crate::error::ApifyClientResult;
11use crate::http_client::{HttpClient, HttpMethod, HttpRequest};
12use crate::models::{
13    RequestQueue, RequestQueueHead, RequestQueueOperationInfo, RequestQueueRequest,
14};
15
16/// Maximum number of requests the API accepts in a single `requests/batch` call. Larger
17/// inputs are split into chunks of this size (matching the reference client).
18const MAX_REQUESTS_PER_BATCH_OPERATION: usize = 25;
19
20/// Appends the array under `key` in `chunk_result` (if present) onto `acc`. Used to merge the
21/// per-chunk `processedRequests` / `unprocessedRequests` arrays of a chunked batch-add.
22fn merge_request_array(
23    acc: &mut Vec<serde_json::Value>,
24    chunk_result: &serde_json::Value,
25    key: &str,
26) {
27    if let Some(items) = chunk_result.get(key).and_then(|v| v.as_array()) {
28        acc.extend(items.iter().cloned());
29    }
30}
31
32/// Options for [`RequestQueueClient::list_requests`].
33///
34/// Covers the spec query parameters of `GET /v2/request-queues/{queueId}/requests`.
35#[derive(Debug, Default, Clone)]
36pub struct ListRequestsOptions {
37    /// Maximum number of requests to return.
38    pub limit: Option<i64>,
39    /// Start listing after this request ID (exclusive).
40    pub exclusive_start_id: Option<String>,
41    /// Opaque pagination cursor returned by a previous call.
42    pub cursor: Option<String>,
43    /// Restrict the returned requests to the given states. The spec defines this as an array of
44    /// the enum values `"locked"` and `"pending"`; multiple values are sent comma-joined (matching
45    /// the JS reference, which serializes `filter: Array<'locked' | 'pending'>` via `join(',')`).
46    pub filter: Option<Vec<String>>,
47}
48
49/// Client for a specific request queue.
50#[derive(Debug, Clone)]
51pub struct RequestQueueClient {
52    ctx: ResourceContext,
53    client_key: Option<String>,
54}
55
56impl RequestQueueClient {
57    pub(crate) fn new(http: HttpClient, base_url: &str, resource_path: &str, id: &str) -> Self {
58        Self {
59            ctx: ResourceContext::single(http, base_url, resource_path, id),
60            client_key: None,
61        }
62    }
63
64    /// Creates an RQ client for a run's default queue (nested path, no ID).
65    pub(crate) fn nested(http: HttpClient, base_url: &str, sub_path: &str) -> Self {
66        Self {
67            ctx: ResourceContext::collection(http, base_url, sub_path),
68            client_key: None,
69        }
70    }
71
72    /// Sets the `clientKey` used to identify this client across requests (for locking).
73    pub fn with_client_key(mut self, client_key: impl Into<String>) -> Self {
74        self.client_key = Some(client_key.into());
75        self
76    }
77
78    fn base_params(&self) -> QueryParams {
79        let mut params = QueryParams::new();
80        params.add_str("clientKey", self.client_key.clone());
81        params
82    }
83
84    /// Fetches the queue metadata, or `None` if it does not exist.
85    pub async fn get(&self) -> ApifyClientResult<Option<RequestQueue>> {
86        get_resource(&self.ctx, None, &QueryParams::new()).await
87    }
88
89    /// Updates the queue metadata (e.g. `name`, `title`).
90    pub async fn update<T: Serialize>(&self, new_fields: &T) -> ApifyClientResult<RequestQueue> {
91        update_resource(&self.ctx, None, new_fields).await
92    }
93
94    /// Deletes the queue.
95    pub async fn delete(&self) -> ApifyClientResult<()> {
96        delete_resource(&self.ctx, None).await
97    }
98
99    /// Lists requests from the head of the queue (without locking them).
100    pub async fn list_head(&self, limit: Option<i64>) -> ApifyClientResult<RequestQueueHead> {
101        let mut params = self.base_params();
102        params.add_int("limit", limit);
103        get_resource_required(&self.ctx, Some("head"), &params).await
104    }
105
106    /// Adds a single request to the queue. If `forefront` is true, adds it to the front.
107    pub async fn add_request(
108        &self,
109        request: &RequestQueueRequest,
110        forefront: bool,
111    ) -> ApifyClientResult<RequestQueueOperationInfo> {
112        let mut params = self.base_params();
113        params.add_bool("forefront", Some(forefront));
114        let body = serde_json::to_vec(request)?;
115        post_with_body(
116            &self.ctx,
117            Some("requests"),
118            &params,
119            Some(body),
120            "application/json",
121        )
122        .await
123    }
124
125    /// Gets a request by ID, or `None` if it does not exist.
126    pub async fn get_request(&self, id: &str) -> ApifyClientResult<Option<RequestQueueRequest>> {
127        get_resource(
128            &self.ctx,
129            Some(&format!("requests/{}", encode_path_segment(id))),
130            &self.base_params(),
131        )
132        .await
133    }
134
135    /// Updates a request (which must include its `id`).
136    pub async fn update_request(
137        &self,
138        request: &RequestQueueRequest,
139        forefront: bool,
140    ) -> ApifyClientResult<RequestQueueOperationInfo> {
141        let id = request.id.clone().ok_or_else(|| {
142            crate::error::ApifyClientError::InvalidArgument(
143                "request.id is required to update a request".to_string(),
144            )
145        })?;
146        let mut params = self.base_params();
147        params.add_bool("forefront", Some(forefront));
148        let url = params.apply_to_url(
149            &self
150                .ctx
151                .url(Some(&format!("requests/{}", encode_path_segment(&id)))),
152        );
153        let body = serde_json::to_vec(request)?;
154        let mut headers = std::collections::HashMap::new();
155        headers.insert("Content-Type".to_string(), "application/json".to_string());
156        let response = self
157            .ctx
158            .http
159            .call(HttpRequest {
160                method: HttpMethod::Put,
161                url,
162                headers,
163                body: Some(body),
164                timeout: crate::clients::base::DEFAULT_REQUEST_TIMEOUT,
165            })
166            .await?;
167        crate::common::parse_data_envelope(&response.body)
168    }
169
170    /// Deletes a request by ID.
171    pub async fn delete_request(&self, id: &str) -> ApifyClientResult<()> {
172        let params = self.base_params();
173        let url = params.apply_to_url(
174            &self
175                .ctx
176                .url(Some(&format!("requests/{}", encode_path_segment(id)))),
177        );
178        self.ctx
179            .http
180            .call(HttpRequest {
181                method: HttpMethod::Delete,
182                url,
183                headers: Default::default(),
184                body: None,
185                timeout: crate::clients::base::DEFAULT_REQUEST_TIMEOUT,
186            })
187            .await?;
188        Ok(())
189    }
190
191    /// Lists and locks requests from the head of the queue for `lock_secs` seconds.
192    pub async fn list_and_lock_head(
193        &self,
194        lock_secs: i64,
195        limit: Option<i64>,
196    ) -> ApifyClientResult<serde_json::Value> {
197        let mut params = self.base_params();
198        params
199            .add_int("lockSecs", Some(lock_secs))
200            .add_int("limit", limit);
201        post_action(&self.ctx, Some("head/lock"), &params, None, None).await
202    }
203
204    /// Adds multiple requests to the queue, automatically splitting the input into chunks of
205    /// at most [`MAX_REQUESTS_PER_BATCH_OPERATION`] requests per API call (the API rejects
206    /// larger batches). The per-chunk responses are merged into a single result whose
207    /// `processedRequests` / `unprocessedRequests` arrays concatenate every chunk's, matching
208    /// the reference client's client-side chunking.
209    pub async fn batch_add_requests(
210        &self,
211        requests: &[RequestQueueRequest],
212        forefront: bool,
213    ) -> ApifyClientResult<serde_json::Value> {
214        let mut processed: Vec<serde_json::Value> = Vec::new();
215        let mut unprocessed: Vec<serde_json::Value> = Vec::new();
216
217        for chunk in requests.chunks(MAX_REQUESTS_PER_BATCH_OPERATION) {
218            let chunk_result = self.batch_add_chunk(chunk, forefront).await?;
219            merge_request_array(&mut processed, &chunk_result, "processedRequests");
220            merge_request_array(&mut unprocessed, &chunk_result, "unprocessedRequests");
221        }
222
223        Ok(serde_json::json!({
224            "processedRequests": processed,
225            "unprocessedRequests": unprocessed,
226        }))
227    }
228
229    /// Posts a single chunk of requests (at most [`MAX_REQUESTS_PER_BATCH_OPERATION`]).
230    async fn batch_add_chunk(
231        &self,
232        requests: &[RequestQueueRequest],
233        forefront: bool,
234    ) -> ApifyClientResult<serde_json::Value> {
235        let mut params = self.base_params();
236        params.add_bool("forefront", Some(forefront));
237        let body = serde_json::to_vec(requests)?;
238        post_with_body(
239            &self.ctx,
240            Some("requests/batch"),
241            &params,
242            Some(body),
243            "application/json",
244        )
245        .await
246    }
247
248    /// Deletes multiple requests in a single batch operation.
249    pub async fn batch_delete_requests<T: Serialize>(
250        &self,
251        requests: &[T],
252    ) -> ApifyClientResult<serde_json::Value> {
253        delete_with_body(
254            &self.ctx,
255            Some("requests/batch"),
256            &self.base_params(),
257            &requests,
258        )
259        .await
260    }
261
262    /// Lists requests in the queue.
263    ///
264    /// Supports pagination via `limit`/`exclusive_start_id` and the spec's `cursor`/`filter`
265    /// parameters (see [`ListRequestsOptions`]).
266    pub async fn list_requests(
267        &self,
268        options: ListRequestsOptions,
269    ) -> ApifyClientResult<serde_json::Value> {
270        let mut params = self.base_params();
271        params
272            .add_int("limit", options.limit)
273            .add_str("exclusiveStartId", options.exclusive_start_id)
274            .add_str("cursor", options.cursor)
275            .add_csv("filter", options.filter.as_deref());
276        get_resource_required(&self.ctx, Some("requests"), &params).await
277    }
278
279    /// Prolongs the lock on a request for another `lock_secs` seconds.
280    ///
281    /// If `forefront` is `true`, the request moves to the front of the queue when its lock
282    /// later expires.
283    pub async fn prolong_request_lock(
284        &self,
285        id: &str,
286        lock_secs: i64,
287        forefront: bool,
288    ) -> ApifyClientResult<serde_json::Value> {
289        let mut params = self.base_params();
290        params
291            .add_int("lockSecs", Some(lock_secs))
292            .add_bool("forefront", Some(forefront));
293        let url = params.apply_to_url(
294            &self
295                .ctx
296                .url(Some(&format!("requests/{}/lock", encode_path_segment(id)))),
297        );
298        let response = self
299            .ctx
300            .http
301            .call(HttpRequest {
302                method: HttpMethod::Put,
303                url,
304                headers: Default::default(),
305                body: None,
306                timeout: crate::clients::base::MEDIUM_REQUEST_TIMEOUT,
307            })
308            .await?;
309        crate::common::parse_data_envelope(&response.body)
310    }
311
312    /// Releases the lock on a request so other clients can process it.
313    ///
314    /// If `forefront` is `true`, the request moves to the front of the queue.
315    pub async fn delete_request_lock(&self, id: &str, forefront: bool) -> ApifyClientResult<()> {
316        let mut params = self.base_params();
317        params.add_bool("forefront", Some(forefront));
318        let url = params.apply_to_url(
319            &self
320                .ctx
321                .url(Some(&format!("requests/{}/lock", encode_path_segment(id)))),
322        );
323        self.ctx
324            .http
325            .call(HttpRequest {
326                method: HttpMethod::Delete,
327                url,
328                headers: Default::default(),
329                body: None,
330                timeout: crate::clients::base::SMALL_REQUEST_TIMEOUT,
331            })
332            .await?;
333        Ok(())
334    }
335
336    /// Lazily paginates over all requests in the queue, fetching pages on demand.
337    ///
338    /// Returns a [`RequestQueueRequestsIterator`]; call its `next()` to get one request at a
339    /// time. Pagination uses the API's opaque `nextCursor` token: the first page may be
340    /// anchored with `exclusiveStartId`, but every subsequent page is fetched with `cursor`
341    /// (matching the JS reference). `cursor` and `exclusiveStartId` are mutually exclusive.
342    pub fn paginate_requests(&self, page_limit: Option<i64>) -> RequestQueueRequestsIterator {
343        RequestQueueRequestsIterator {
344            client: self.clone(),
345            page_limit,
346            buffer: std::collections::VecDeque::new(),
347            next_cursor: None,
348            exhausted: false,
349        }
350    }
351
352    /// Unlocks all requests currently locked by this client (identified by `client_key`).
353    pub async fn unlock_requests(&self) -> ApifyClientResult<serde_json::Value> {
354        post_action(
355            &self.ctx,
356            Some("requests/unlock"),
357            &self.base_params(),
358            None,
359            None,
360        )
361        .await
362    }
363}
364
365/// A lazy, page-fetching iterator over the requests in a queue.
366///
367/// Created by [`RequestQueueClient::paginate_requests`]. Each call to [`next`](Self::next)
368/// returns the next request, fetching another page from the API when the local buffer is
369/// exhausted, until all requests have been yielded.
370pub struct RequestQueueRequestsIterator {
371    client: RequestQueueClient,
372    page_limit: Option<i64>,
373    buffer: std::collections::VecDeque<RequestQueueRequest>,
374    /// Opaque pagination token returned by the previous page, fed back as `cursor`.
375    next_cursor: Option<String>,
376    exhausted: bool,
377}
378
379impl RequestQueueRequestsIterator {
380    /// Returns the next request, or `None` when all requests have been yielded.
381    pub async fn next(&mut self) -> ApifyClientResult<Option<RequestQueueRequest>> {
382        if let Some(item) = self.buffer.pop_front() {
383            return Ok(Some(item));
384        }
385        if self.exhausted {
386            return Ok(None);
387        }
388
389        // The first page may be anchored by exclusiveStartId; every later page is fetched
390        // with the opaque `cursor` token (mutually exclusive with exclusiveStartId), matching
391        // the JS reference. Here we only ever paginate from the queue head, so the first page
392        // uses neither and subsequent pages use `cursor`.
393        let page = self
394            .client
395            .list_requests(ListRequestsOptions {
396                limit: self.page_limit,
397                cursor: self.next_cursor.clone(),
398                ..Default::default()
399            })
400            .await?;
401
402        // Parse the items and the next cursor from the (untyped) page.
403        let items: Vec<RequestQueueRequest> = page
404            .get("items")
405            .map(|v| serde_json::from_value(v.clone()))
406            .transpose()?
407            .unwrap_or_default();
408
409        if items.is_empty() {
410            self.exhausted = true;
411            return Ok(None);
412        }
413
414        // Advance the cursor; stop when the API stops returning one.
415        match page.get("nextCursor").and_then(|v| v.as_str()) {
416            Some(cursor) if !cursor.is_empty() => self.next_cursor = Some(cursor.to_string()),
417            _ => self.exhausted = true,
418        }
419
420        self.buffer.extend(items);
421        Ok(self.buffer.pop_front())
422    }
423}
apify_client/clients/request_queue.rs

apify_client/clients/
request_queue.rs
