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        // Fall back to `0` ("total unknown"), never `count`: a total equal to the items already
214        // returned would look complete and stop iteration after page one, dropping later items.
215        // `0` routes iteration to the short-page/empty-page backstop, which walks every page.
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
229        Ok(PaginationList {
230            total,
231            offset,
232            limit,
233            count,
234            desc: options.desc.unwrap_or(false),
235            items,
236        })
237    }
238
239    /// Lazily iterates over all items in the dataset, fetching pages on demand.
240    ///
241    /// The idiomatic-Rust counterpart of the reference client's async-iterable
242    /// `listItems`/`iterateItems`: yields one deserialized item of type `T` at a time,
243    /// transparently paging. The caller's `options.limit` caps the total number of items yielded
244    /// (unset = all); use [`ListIterator::with_chunk_size`] to control the per-page fetch size.
245    ///
246    /// Server-side filters (`skip_empty`/`clean`/`skip_hidden`) are forwarded on every page
247    /// request. Paging advances the offset by the number of items each page returns, exactly
248    /// like the reference JavaScript client (`currentOffset += items.length`). Because the offset
249    /// advances by the post-filter count rather than the raw window size, filtered iteration is
250    /// not exact — this matches the reference client, and it can distort results two ways:
251    ///
252    /// - Duplicates: when a page's filtered count is smaller than its raw window, the next page
253    ///   starts at an offset that overlaps the previous window, so some items are yielded more
254    ///   than once.
255    /// - Dropped items: when a filter removes *every* item in a raw window, the page comes back
256    ///   with no items; iteration treats that empty page as the end of the dataset (the empty-page
257    ///   backstop in [`ListIterator`]) and stops, even though unfiltered items still exist at
258    ///   higher offsets.
259    ///
260    /// If you need every filtered item exactly once, apply the filter client-side over an
261    /// unfiltered iteration instead.
262    pub fn iterate_items<T: DeserializeOwned + Send + 'static>(
263        &self,
264        options: DatasetListItemsOptions,
265    ) -> ListIterator<T> {
266        let client = self.clone();
267        let start = options.offset.unwrap_or(0);
268        let total_limit = options.limit;
269        ListIterator::new(
270            start,
271            total_limit,
272            Box::new(move |offset, page_limit| {
273                let client = client.clone();
274                let mut options = options.clone();
275                options.offset = Some(offset);
276                options.limit = page_limit;
277                Box::pin(async move { client.list_items::<T>(options).await })
278            }),
279        )
280    }
281
282    /// Downloads dataset items serialized in the given `format`, returning the raw bytes.
283    ///
284    /// Unlike [`list_items`](Self::list_items), which returns parsed items, this returns the
285    /// items already serialized to JSON, CSV, XLSX, XML, RSS or HTML — useful for exporting.
286    /// Use [`DatasetDownloadOptions`] to control export-specific behaviour (BOM, CSV
287    /// delimiter/header, XML element names, attachment disposition).
288    pub async fn download_items(
289        &self,
290        format: DownloadItemsFormat,
291        options: DatasetDownloadOptions,
292    ) -> ApifyClientResult<Vec<u8>> {
293        let mut params = QueryParams::new();
294        params.add_str("format", Some(format.as_str()));
295        options.apply(&mut params);
296        let url = params.apply_to_url(&self.ctx.url(Some("items")));
297        let response = self
298            .ctx
299            .http
300            .call(HttpRequest {
301                method: HttpMethod::Get,
302                url,
303                headers: Default::default(),
304                body: None,
305                timeout: crate::clients::base::DEFAULT_REQUEST_TIMEOUT,
306            })
307            .await?;
308        Ok(response.body)
309    }
310
311    /// Pushes one or more items to the dataset.
312    ///
313    /// `items` must serialize to a JSON object or an array of objects.
314    pub async fn push_items<T: Serialize>(&self, items: &T) -> ApifyClientResult<()> {
315        let body = serde_json::to_vec(items)?;
316        let url = self.ctx.url(Some("items"));
317        let mut headers = std::collections::HashMap::new();
318        headers.insert(
319            "Content-Type".to_string(),
320            "application/json; charset=utf-8".to_string(),
321        );
322        self.ctx
323            .http
324            .call(HttpRequest {
325                method: HttpMethod::Post,
326                url,
327                headers,
328                body: Some(body),
329                timeout: crate::clients::base::DEFAULT_REQUEST_TIMEOUT,
330            })
331            .await?;
332        Ok(())
333    }
334
335    /// Builds a public URL for downloading this dataset's items.
336    ///
337    /// Mirrors the reference client's `createItemsPublicUrl`: it fetches the dataset, and if
338    /// the dataset exposes a URL-signing secret key (i.e. it is private), appends an
339    /// HMAC-SHA256 `signature` so the URL grants access without an API token. `expires_in_secs`
340    /// optionally bounds the validity of a signed URL. The URL is built from the configured
341    /// public base URL.
342    pub async fn create_items_public_url(
343        &self,
344        options: DatasetListItemsOptions,
345        expires_in_secs: Option<i64>,
346    ) -> ApifyClientResult<String> {
347        let mut params = QueryParams::new();
348        options.apply(&mut params);
349
350        if let Some(dataset) = self.get().await? {
351            if let Some(secret) = dataset
352                .extra
353                .get("urlSigningSecretKey")
354                .and_then(|v| v.as_str())
355            {
356                let signature = sign_storage_content(secret, &dataset.id, expires_in_secs);
357                params.add_str("signature", Some(signature));
358            }
359        }
360        Ok(params.apply_to_url(&self.ctx.public_url(Some("items"))))
361    }
362
363    /// Returns statistical information about the dataset, or `None` if unavailable.
364    pub async fn get_statistics(&self) -> ApifyClientResult<Option<Value>> {
365        let result: ApifyClientResult<Value> = async {
366            let response = self
367                .ctx
368                .http
369                .call(HttpRequest {
370                    method: HttpMethod::Get,
371                    url: self.ctx.url(Some("statistics")),
372                    headers: Default::default(),
373                    body: None,
374                    timeout: crate::clients::base::DEFAULT_REQUEST_TIMEOUT,
375                })
376                .await?;
377            parse_data_envelope(&response.body)
378        }
379        .await;
380        crate::common::catch_not_found(result)
381    }
382}
apify_client/clients/dataset.rs

apify_client/clients/
dataset.rs
