apify_client/clients/
dataset.rs

1//! Client for a single dataset (`/v2/datasets/{datasetId}` and run-nested variants).
2
3use serde::de::DeserializeOwned;
4use serde::Serialize;
5use serde_json::Value;
6
7use crate::clients::base::{delete_resource, get_resource, update_resource, ResourceContext};
8use crate::clients::pagination::ListIterator;
9use crate::common::{parse_data_envelope, sign_storage_content, PaginationList, QueryParams};
10use crate::error::ApifyClientResult;
11use crate::http_client::{HttpClient, HttpMethod, HttpRequest};
12use crate::models::Dataset;
13
14/// Options for listing or downloading dataset items.
15///
16/// Covers the filtering, projection and transformation parameters of
17/// `GET /v2/datasets/{datasetId}/items`.
18#[derive(Debug, Default, Clone)]
19pub struct DatasetListItemsOptions {
20    /// Number of items to skip.
21    pub offset: Option<i64>,
22    /// Maximum number of items to return.
23    pub limit: Option<i64>,
24    /// Return items newest-first.
25    pub desc: Option<bool>,
26    /// Only include these fields.
27    pub fields: Option<Vec<String>>,
28    /// Positionally renames the fields selected by `fields` in the output (requires `fields`
29    /// to be set). The i-th name here becomes the output name of the i-th `fields` entry.
30    pub output_fields: Option<Vec<String>>,
31    /// Exclude these fields.
32    pub omit: Option<Vec<String>>,
33    /// Skip empty items.
34    pub skip_empty: Option<bool>,
35    /// Skip hidden fields (those starting with `#`).
36    pub skip_hidden: Option<bool>,
37    /// Only return clean (non-empty, non-hidden) items.
38    pub clean: Option<bool>,
39    /// Unwind these fields (each array element becomes a separate item).
40    pub unwind: Option<Vec<String>>,
41    /// Flatten these nested fields into dot-notation keys.
42    pub flatten: Option<Vec<String>>,
43    /// Use a predefined dataset view for field selection.
44    pub view: Option<String>,
45    /// Return simplified (flattened, cleaned) items.
46    pub simplified: Option<bool>,
47    /// Skip items that come from failed pages.
48    pub skip_failed_pages: Option<bool>,
49    /// Pre-shared URL signature granting access to a private dataset without an API token.
50    pub signature: Option<String>,
51}
52
53impl DatasetListItemsOptions {
54    fn apply(&self, params: &mut QueryParams) {
55        params
56            .add_int("offset", self.offset)
57            .add_int("limit", self.limit)
58            .add_bool("desc", self.desc)
59            .add_csv("fields", self.fields.as_deref())
60            .add_csv("outputFields", self.output_fields.as_deref())
61            .add_csv("omit", self.omit.as_deref())
62            .add_bool("skipEmpty", self.skip_empty)
63            .add_bool("skipHidden", self.skip_hidden)
64            .add_bool("clean", self.clean)
65            .add_csv("unwind", self.unwind.as_deref())
66            .add_csv("flatten", self.flatten.as_deref())
67            .add_str("view", self.view.clone())
68            .add_bool("simplified", self.simplified)
69            .add_bool("skipFailedPages", self.skip_failed_pages)
70            .add_str("signature", self.signature.clone());
71    }
72}
73
74/// Output formats supported by [`DatasetClient::download_items`].
75#[derive(Debug, Clone, Copy, PartialEq, Eq)]
76pub enum DownloadItemsFormat {
77    /// JSON array.
78    Json,
79    /// Newline-delimited JSON.
80    Jsonl,
81    /// Comma-separated values.
82    Csv,
83    /// Microsoft Excel (XLSX).
84    Xlsx,
85    /// XML.
86    Xml,
87    /// RSS feed.
88    Rss,
89    /// HTML table.
90    Html,
91}
92
93impl DownloadItemsFormat {
94    fn as_str(&self) -> &'static str {
95        match self {
96            DownloadItemsFormat::Json => "json",
97            DownloadItemsFormat::Jsonl => "jsonl",
98            DownloadItemsFormat::Csv => "csv",
99            DownloadItemsFormat::Xlsx => "xlsx",
100            DownloadItemsFormat::Xml => "xml",
101            DownloadItemsFormat::Rss => "rss",
102            DownloadItemsFormat::Html => "html",
103        }
104    }
105}
106
107/// Format-specific options for [`DatasetClient::download_items`], on top of the shared
108/// filtering/projection options.
109#[derive(Debug, Default, Clone)]
110pub struct DatasetDownloadOptions {
111    /// Shared item filtering/projection options.
112    pub items: DatasetListItemsOptions,
113    /// Set `Content-Disposition: attachment` on the response.
114    pub attachment: Option<bool>,
115    /// Prepend a UTF-8 BOM (useful for Excel-compatible CSV).
116    pub bom: Option<bool>,
117    /// CSV field delimiter (default `,`).
118    pub delimiter: Option<String>,
119    /// Omit the CSV header row.
120    pub skip_header_row: Option<bool>,
121    /// Name of the root XML element (default `items`).
122    pub xml_root: Option<String>,
123    /// Name of the per-item XML element (default `item`).
124    pub xml_row: Option<String>,
125    /// Title to use for RSS/Atom feed exports.
126    pub feed_title: Option<String>,
127    /// Description to use for RSS/Atom feed exports.
128    pub feed_description: Option<String>,
129}
130
131impl DatasetDownloadOptions {
132    fn apply(&self, params: &mut QueryParams) {
133        self.items.apply(params);
134        params
135            .add_bool("attachment", self.attachment)
136            .add_bool("bom", self.bom)
137            .add_str("delimiter", self.delimiter.clone())
138            .add_bool("skipHeaderRow", self.skip_header_row)
139            .add_str("xmlRoot", self.xml_root.clone())
140            .add_str("xmlRow", self.xml_row.clone())
141            .add_str("feedTitle", self.feed_title.clone())
142            .add_str("feedDescription", self.feed_description.clone());
143    }
144}
145
146/// Client for a specific dataset.
147#[derive(Debug, Clone)]
148pub struct DatasetClient {
149    ctx: ResourceContext,
150}
151
152impl DatasetClient {
153    pub(crate) fn new(http: HttpClient, base_url: &str, resource_path: &str, id: &str) -> Self {
154        Self {
155            ctx: ResourceContext::single(http, base_url, resource_path, id),
156        }
157    }
158
159    /// Creates a dataset client for a run's default dataset (no ID; nested path only).
160    pub(crate) fn nested(http: HttpClient, base_url: &str, sub_path: &str) -> Self {
161        Self {
162            ctx: ResourceContext::collection(http, base_url, sub_path),
163        }
164    }
165
166    /// Sets the public origin used when building shareable URLs.
167    pub(crate) fn with_public_base(mut self, public_base_url: &str) -> Self {
168        self.ctx = self.ctx.with_public_origin(public_base_url);
169        self
170    }
171
172    /// Fetches the dataset metadata, or `None` if it does not exist.
173    pub async fn get(&self) -> ApifyClientResult<Option<Dataset>> {
174        get_resource(&self.ctx, None, &QueryParams::new()).await
175    }
176
177    /// Updates the dataset metadata (e.g. `name`, `title`).
178    pub async fn update<T: Serialize>(&self, new_fields: &T) -> ApifyClientResult<Dataset> {
179        update_resource(&self.ctx, None, new_fields).await
180    }
181
182    /// Deletes the dataset.
183    pub async fn delete(&self) -> ApifyClientResult<()> {
184        delete_resource(&self.ctx, None).await
185    }
186
187    /// Lists items from the dataset.
188    ///
189    /// The dataset items endpoint returns a bare JSON array (not a `data` envelope) and
190    /// reports pagination via `X-Apify-Pagination-*` headers, which are surfaced in the
191    /// returned [`PaginationList`].
192    pub async fn list_items<T: DeserializeOwned>(
193        &self,
194        options: DatasetListItemsOptions,
195    ) -> ApifyClientResult<PaginationList<T>> {
196        let mut params = QueryParams::new();
197        options.apply(&mut params);
198        let url = params.apply_to_url(&self.ctx.url(Some("items")));
199        let response = self
200            .ctx
201            .http
202            .call(HttpRequest {
203                method: HttpMethod::Get,
204                url,
205                headers: Default::default(),
206                body: None,
207                timeout: crate::clients::base::DEFAULT_REQUEST_TIMEOUT,
208            })
209            .await?;
210
211        let items: Vec<T> = serde_json::from_slice(&response.body)?;
212        let count = items.len() as i64;
213        // `PaginationList::total` is informational only here (`ListIterator` never consults it,
214        // see `scanned` below); falling back to `0` rather than `count` just avoids implying a
215        // precise total when the API did not report one.
216        let total = response
217            .header("x-apify-pagination-total")
218            .and_then(|v| v.parse().ok())
219            .unwrap_or(0);
220        let offset = response
221            .header("x-apify-pagination-offset")
222            .and_then(|v| v.parse().ok())
223            .unwrap_or(0);
224        let limit = response
225            .header("x-apify-pagination-limit")
226            .and_then(|v| v.parse().ok())
227            .unwrap_or(count);
228        // The number of rows scanned to produce this page, when the API reports it separately
229        // from the item count (see `PaginationList::scanned`): a filter can scan more rows than
230        // it returns, and `unwind` can return more items than rows scanned.
231        let scanned = response
232            .header("x-apify-pagination-count")
233            .and_then(|v| v.parse().ok());
234
235        Ok(PaginationList {
236            total,
237            offset,
238            limit,
239            count,
240            desc: options.desc.unwrap_or(false),
241            items,
242            scanned,
243        })
244    }
245
246    /// Lazily iterates over all items in the dataset, fetching pages on demand.
247    ///
248    /// The idiomatic-Rust counterpart of the reference client's async-iterable
249    /// `listItems`/`iterateItems`: yields one deserialized item of type `T` at a time,
250    /// transparently paging. The caller's `options.limit` caps the total number of items yielded
251    /// (unset = all); use [`ListIterator::with_chunk_size`] to control the per-page fetch size.
252    ///
253    /// Server-side filters (`skip_empty`/`clean`/`skip_hidden`) and `unwind` are forwarded on
254    /// every page request. Paging advances the offset by the number of underlying rows each page
255    /// *scanned* (the `x-apify-pagination-count` response header), not by `items.len()`, matching
256    /// the reference JavaScript client: a filter can scan more rows than it returns, and `unwind`
257    /// can return more items than rows scanned, so advancing by the returned count alone would
258    /// re-scan or skip rows. The listing's reported `total` is never consulted, so a dataset a
259    /// still-running Actor keeps pushing to is read to the end: iteration keeps requesting pages
260    /// until one scans no rows (one extra, empty request past the last page with data), rather
261    /// than stopping at the size the dataset had when the loop started.
262    ///
263    /// If `x-apify-pagination-count` is ever absent, the scanned count falls back to `items.len()`.
264    pub fn iterate_items<T: DeserializeOwned + Send + 'static>(
265        &self,
266        options: DatasetListItemsOptions,
267    ) -> ListIterator<T> {
268        let client = self.clone();
269        let start = options.offset.unwrap_or(0);
270        let total_limit = options.limit;
271        ListIterator::new(
272            start,
273            total_limit,
274            Box::new(move |offset, page_limit| {
275                let client = client.clone();
276                let mut options = options.clone();
277                options.offset = Some(offset);
278                options.limit = page_limit;
279                Box::pin(async move { client.list_items::<T>(options).await })
280            }),
281        )
282    }
283
284    /// Downloads dataset items serialized in the given `format`, returning the raw bytes.
285    ///
286    /// Unlike [`list_items`](Self::list_items), which returns parsed items, this returns the
287    /// items already serialized to JSON, CSV, XLSX, XML, RSS or HTML — useful for exporting.
288    /// Use [`DatasetDownloadOptions`] to control export-specific behaviour (BOM, CSV
289    /// delimiter/header, XML element names, attachment disposition).
290    pub async fn download_items(
291        &self,
292        format: DownloadItemsFormat,
293        options: DatasetDownloadOptions,
294    ) -> ApifyClientResult<Vec<u8>> {
295        let mut params = QueryParams::new();
296        params.add_str("format", Some(format.as_str()));
297        options.apply(&mut params);
298        let url = params.apply_to_url(&self.ctx.url(Some("items")));
299        let response = self
300            .ctx
301            .http
302            .call(HttpRequest {
303                method: HttpMethod::Get,
304                url,
305                headers: Default::default(),
306                body: None,
307                timeout: crate::clients::base::DEFAULT_REQUEST_TIMEOUT,
308            })
309            .await?;
310        Ok(response.body)
311    }
312
313    /// Pushes one or more items to the dataset.
314    ///
315    /// `items` must serialize to a JSON object or an array of objects.
316    pub async fn push_items<T: Serialize>(&self, items: &T) -> ApifyClientResult<()> {
317        let body = serde_json::to_vec(items)?;
318        let url = self.ctx.url(Some("items"));
319        let mut headers = std::collections::HashMap::new();
320        headers.insert(
321            "Content-Type".to_string(),
322            "application/json; charset=utf-8".to_string(),
323        );
324        self.ctx
325            .http
326            .call(HttpRequest {
327                method: HttpMethod::Post,
328                url,
329                headers,
330                body: Some(body),
331                timeout: crate::clients::base::DEFAULT_REQUEST_TIMEOUT,
332            })
333            .await?;
334        Ok(())
335    }
336
337    /// Builds a public URL for downloading this dataset's items.
338    ///
339    /// Mirrors the reference client's `createItemsPublicUrl`: it fetches the dataset, and if
340    /// the dataset exposes a URL-signing secret key (i.e. it is private), appends an
341    /// HMAC-SHA256 `signature` so the URL grants access without an API token. `expires_in_secs`
342    /// optionally bounds the validity of a signed URL. The URL is built from the configured
343    /// public base URL. The served items are JSON; to request another export format, use
344    /// [`create_items_public_url_with_format`](Self::create_items_public_url_with_format).
345    pub async fn create_items_public_url(
346        &self,
347        options: DatasetListItemsOptions,
348        expires_in_secs: Option<i64>,
349    ) -> ApifyClientResult<String> {
350        self.create_items_public_url_with_format(options, expires_in_secs, None)
351            .await
352    }
353
354    /// Like [`create_items_public_url`](Self::create_items_public_url), but lets the caller
355    /// choose the `format` the items are served in (JSON, CSV, XLSX, ...) via a `format` query
356    /// parameter on the generated URL. `None` defaults to `json`, matching the endpoint's own
357    /// default.
358    pub async fn create_items_public_url_with_format(
359        &self,
360        options: DatasetListItemsOptions,
361        expires_in_secs: Option<i64>,
362        format: Option<DownloadItemsFormat>,
363    ) -> ApifyClientResult<String> {
364        let mut params = QueryParams::new();
365        options.apply(&mut params);
366        params.add_str("format", format.map(|f| f.as_str().to_string()));
367
368        if let Some(dataset) = self.get().await? {
369            if let Some(secret) = dataset
370                .extra
371                .get("urlSigningSecretKey")
372                .and_then(|v| v.as_str())
373            {
374                let signature = sign_storage_content(secret, &dataset.id, expires_in_secs);
375                params.add_str("signature", Some(signature));
376            }
377        }
378        Ok(params.apply_to_url(&self.ctx.public_url(Some("items"))))
379    }
380
381    /// Returns statistical information about the dataset, or `None` if unavailable.
382    pub async fn get_statistics(&self) -> ApifyClientResult<Option<Value>> {
383        let result: ApifyClientResult<Value> = async {
384            let response = self
385                .ctx
386                .http
387                .call(HttpRequest {
388                    method: HttpMethod::Get,
389                    url: self.ctx.url(Some("statistics")),
390                    headers: Default::default(),
391                    body: None,
392                    timeout: crate::clients::base::DEFAULT_REQUEST_TIMEOUT,
393                })
394                .await?;
395            parse_data_envelope(&response.body)
396        }
397        .await;
398        crate::common::catch_not_found(result)
399    }
400}
apify_client/clients/dataset.rs

apify_client/clients/
dataset.rs
