meilisearch_sdk/
tasks.rs

1use serde::{Deserialize, Deserializer, Serialize};
2use serde_json::{Map, Value};
3use std::time::Duration;
4use time::OffsetDateTime;
5
6use crate::{
7    client::Client, client::SwapIndexes, errors::Error, errors::MeilisearchError, indexes::Index,
8    request::HttpClient, settings::Settings, task_info::TaskInfo,
9};
10
11#[derive(Debug, Clone, Deserialize)]
12#[serde(rename_all = "camelCase", tag = "type")]
13pub enum TaskType {
14    Customs,
15    DocumentAdditionOrUpdate {
16        details: Option<DocumentAdditionOrUpdate>,
17    },
18    DocumentDeletion {
19        details: Option<DocumentDeletion>,
20    },
21    IndexCreation {
22        details: Option<IndexCreation>,
23    },
24    IndexUpdate {
25        details: Option<IndexUpdate>,
26    },
27    IndexDeletion {
28        details: Option<IndexDeletion>,
29    },
30    SettingsUpdate {
31        details: Box<Option<Settings>>,
32    },
33    DumpCreation {
34        details: Option<DumpCreation>,
35    },
36    IndexSwap {
37        details: Option<IndexSwap>,
38    },
39    NetworkTopologyChange {
40        #[serde(skip_serializing_if = "Option::is_none")]
41        details: Option<NetworkTopologyChangeDetails>,
42    },
43    TaskCancelation {
44        details: Option<TaskCancelation>,
45    },
46    TaskDeletion {
47        details: Option<TaskDeletion>,
48    },
49    SnapshotCreation {
50        details: Option<SnapshotCreation>,
51    },
52    IndexCompaction {
53        details: Option<IndexCompaction>,
54    },
55}
56
57#[derive(Debug, Clone, Deserialize)]
58pub struct TasksResults {
59    pub results: Vec<Task>,
60    pub total: u64,
61    pub limit: u32,
62    pub from: Option<u32>,
63    pub next: Option<u32>,
64}
65
66#[derive(Debug, Clone, Deserialize)]
67#[serde(rename_all = "camelCase")]
68pub struct DocumentAdditionOrUpdate {
69    pub indexed_documents: Option<usize>,
70    pub received_documents: usize,
71}
72
73#[derive(Debug, Clone, Deserialize)]
74#[serde(rename_all = "camelCase")]
75pub struct DocumentDeletion {
76    pub provided_ids: Option<usize>,
77    pub deleted_documents: Option<usize>,
78    pub original_filter: Option<String>,
79}
80
81#[derive(Debug, Clone, Deserialize)]
82#[serde(rename_all = "camelCase")]
83pub struct IndexCreation {
84    pub primary_key: Option<String>,
85}
86
87#[derive(Debug, Clone, Deserialize)]
88#[serde(rename_all = "camelCase")]
89pub struct IndexUpdate {
90    pub primary_key: Option<String>,
91}
92
93#[derive(Debug, Clone, Deserialize)]
94#[serde(rename_all = "camelCase")]
95pub struct IndexDeletion {
96    pub deleted_documents: Option<usize>,
97}
98
99#[derive(Debug, Clone, Deserialize)]
100#[serde(rename_all = "camelCase")]
101pub struct SnapshotCreation {}
102
103#[derive(Debug, Clone, Deserialize)]
104#[serde(rename_all = "camelCase")]
105pub struct IndexCompaction {}
106
107#[derive(Debug, Clone, Deserialize)]
108#[serde(rename_all = "camelCase")]
109pub struct NetworkTopologyChangeDetails {
110    #[serde(flatten)]
111    pub info: Map<String, Value>,
112}
113
114#[derive(Debug, Clone, Deserialize)]
115#[serde(rename_all = "camelCase")]
116pub struct DumpCreation {
117    pub dump_uid: Option<String>,
118}
119
120#[derive(Debug, Clone, Deserialize)]
121#[serde(rename_all = "camelCase")]
122pub struct IndexSwap {
123    pub swaps: Vec<SwapIndexes>,
124}
125
126#[derive(Debug, Clone, Deserialize)]
127#[serde(rename_all = "camelCase")]
128pub struct TaskCancelation {
129    pub matched_tasks: usize,
130    pub canceled_tasks: Option<usize>,
131    pub original_filter: String,
132}
133
134#[derive(Debug, Clone, Deserialize)]
135#[serde(rename_all = "camelCase")]
136pub struct TaskDeletion {
137    pub matched_tasks: usize,
138    pub deleted_tasks: Option<usize>,
139    pub original_filter: String,
140}
141
142#[derive(Deserialize, Debug, Clone)]
143#[serde(rename_all = "camelCase")]
144pub struct FailedTask {
145    pub error: MeilisearchError,
146    #[serde(flatten)]
147    pub task: SucceededTask,
148}
149
150impl AsRef<u32> for FailedTask {
151    fn as_ref(&self) -> &u32 {
152        &self.task.uid
153    }
154}
155
156fn deserialize_duration<'de, D>(deserializer: D) -> Result<Duration, D::Error>
157where
158    D: Deserializer<'de>,
159{
160    let s = String::deserialize(deserializer)?;
161    let iso_duration = iso8601::duration(&s).map_err(serde::de::Error::custom)?;
162    Ok(iso_duration.into())
163}
164
165#[derive(Deserialize, Debug, Clone)]
166#[serde(rename_all = "camelCase")]
167pub struct SucceededTask {
168    #[serde(deserialize_with = "deserialize_duration")]
169    pub duration: Duration,
170    #[serde(with = "time::serde::rfc3339")]
171    pub enqueued_at: OffsetDateTime,
172    #[serde(with = "time::serde::rfc3339")]
173    pub started_at: OffsetDateTime,
174    #[serde(with = "time::serde::rfc3339")]
175    pub finished_at: OffsetDateTime,
176    pub canceled_by: Option<usize>,
177    pub index_uid: Option<String>,
178    pub error: Option<MeilisearchError>,
179    /// Remotes object returned by the server for this task (present since Meilisearch 1.19)
180    #[serde(skip_serializing_if = "Option::is_none")]
181    pub remotes: Option<Map<String, Value>>,
182    #[serde(flatten)]
183    pub update_type: TaskType,
184    pub uid: u32,
185}
186
187impl AsRef<u32> for SucceededTask {
188    fn as_ref(&self) -> &u32 {
189        &self.uid
190    }
191}
192
193#[derive(Debug, Clone, Deserialize)]
194#[serde(rename_all = "camelCase")]
195pub struct EnqueuedTask {
196    #[serde(with = "time::serde::rfc3339")]
197    pub enqueued_at: OffsetDateTime,
198    pub index_uid: Option<String>,
199    /// Remotes object returned by the server for this enqueued task
200    #[serde(skip_serializing_if = "Option::is_none")]
201    pub remotes: Option<Map<String, Value>>,
202    #[serde(flatten)]
203    pub update_type: TaskType,
204    pub uid: u32,
205}
206
207impl AsRef<u32> for EnqueuedTask {
208    fn as_ref(&self) -> &u32 {
209        &self.uid
210    }
211}
212
213#[derive(Debug, Clone, Deserialize)]
214#[serde(rename_all = "camelCase")]
215pub struct ProcessingTask {
216    #[serde(with = "time::serde::rfc3339")]
217    pub enqueued_at: OffsetDateTime,
218    #[serde(with = "time::serde::rfc3339")]
219    pub started_at: OffsetDateTime,
220    pub index_uid: Option<String>,
221    /// Remotes object returned by the server for this processing task
222    #[serde(skip_serializing_if = "Option::is_none")]
223    pub remotes: Option<Map<String, Value>>,
224    #[serde(flatten)]
225    pub update_type: TaskType,
226    pub uid: u32,
227}
228
229impl AsRef<u32> for ProcessingTask {
230    fn as_ref(&self) -> &u32 {
231        &self.uid
232    }
233}
234
235#[derive(Debug, Clone, Deserialize)]
236#[serde(rename_all = "camelCase", tag = "status")]
237pub enum Task {
238    Enqueued {
239        #[serde(flatten)]
240        content: EnqueuedTask,
241    },
242    Processing {
243        #[serde(flatten)]
244        content: ProcessingTask,
245    },
246    Failed {
247        #[serde(flatten)]
248        content: FailedTask,
249    },
250    Succeeded {
251        #[serde(flatten)]
252        content: SucceededTask,
253    },
254}
255
256impl Task {
257    #[must_use]
258    pub fn get_uid(&self) -> u32 {
259        match self {
260            Self::Enqueued { content } => *content.as_ref(),
261            Self::Processing { content } => *content.as_ref(),
262            Self::Failed { content } => *content.as_ref(),
263            Self::Succeeded { content } => *content.as_ref(),
264        }
265    }
266
267    /// Wait until Meilisearch processes a [Task], and get its status.
268    ///
269    /// `interval` = The frequency at which the server should be polled. **Default = 50ms**
270    ///
271    /// `timeout` = The maximum time to wait for processing to complete. **Default = 5000ms**
272    ///
273    /// If the waited time exceeds `timeout` then an [`Error::Timeout`] will be returned.
274    ///
275    /// See also [`Client::wait_for_task`, `Index::wait_for_task`].
276    ///
277    /// # Example
278    ///
279    /// ```
280    /// # use meilisearch_sdk::{client::*, indexes::*, tasks::Task};
281    /// # use serde::{Serialize, Deserialize};
282    /// #
283    /// # let MEILISEARCH_URL = option_env!("MEILISEARCH_URL").unwrap_or("http://localhost:7700");
284    /// # let MEILISEARCH_API_KEY = option_env!("MEILISEARCH_API_KEY").unwrap_or("masterKey");
285    /// #
286    /// # #[derive(Debug, Serialize, Deserialize, PartialEq)]
287    /// # struct Document {
288    /// #    id: usize,
289    /// #    value: String,
290    /// #    kind: String,
291    /// # }
292    /// #
293    /// #
294    /// # tokio::runtime::Builder::new_current_thread().enable_all().build().unwrap().block_on(async {
295    /// # let client = Client::new(MEILISEARCH_URL, Some(MEILISEARCH_API_KEY)).unwrap();
296    /// let movies = client.index("movies_wait_for_completion");
297    ///
298    /// let status = movies.add_documents(&[
299    ///     Document { id: 0, kind: "title".into(), value: "The Social Network".to_string() },
300    ///     Document { id: 1, kind: "title".into(), value: "Harry Potter and the Sorcerer's Stone".to_string() },
301    /// ], None)
302    ///     .await
303    ///     .unwrap()
304    ///     .wait_for_completion(&client, None, None)
305    ///     .await
306    ///     .unwrap();
307    ///
308    /// assert!(matches!(status, Task::Succeeded { .. }));
309    /// # movies.delete().await.unwrap().wait_for_completion(&client, None, None).await.unwrap();
310    /// # });
311    /// ```
312    pub async fn wait_for_completion<Http: HttpClient>(
313        self,
314        client: &Client<Http>,
315        interval: Option<Duration>,
316        timeout: Option<Duration>,
317    ) -> Result<Self, Error> {
318        client.wait_for_task(self, interval, timeout).await
319    }
320
321    /// Extract the [Index] from a successful `IndexCreation` task.
322    ///
323    /// If the task failed or was not an `IndexCreation` task it returns itself.
324    ///
325    /// # Example
326    ///
327    /// ```
328    /// # use meilisearch_sdk::{client::*, indexes::*};
329    /// #
330    /// # let MEILISEARCH_URL = option_env!("MEILISEARCH_URL").unwrap_or("http://localhost:7700");
331    /// # let MEILISEARCH_API_KEY = option_env!("MEILISEARCH_API_KEY").unwrap_or("masterKey");
332    /// #
333    /// # tokio::runtime::Builder::new_current_thread().enable_all().build().unwrap().block_on(async {
334    /// # // create the client
335    /// # let client = Client::new(MEILISEARCH_URL, Some(MEILISEARCH_API_KEY)).unwrap();
336    /// let task = client.create_index("try_make_index", None).await.unwrap();
337    /// let index = client.wait_for_task(task, None, None).await.unwrap().try_make_index(&client).unwrap();
338    ///
339    /// // and safely access it
340    /// assert_eq!(index.as_ref(), "try_make_index");
341    /// # index.delete().await.unwrap().wait_for_completion(&client, None, None).await.unwrap();
342    /// # });
343    /// ```
344    #[allow(clippy::result_large_err)] // Since `self` has been consumed, this is not an issue
345    pub fn try_make_index<Http: HttpClient>(
346        self,
347        client: &Client<Http>,
348    ) -> Result<Index<Http>, Self> {
349        match self {
350            Self::Succeeded {
351                content:
352                    SucceededTask {
353                        index_uid,
354                        update_type: TaskType::IndexCreation { .. },
355                        ..
356                    },
357            } => Ok(client.index(index_uid.unwrap())),
358            _ => Err(self),
359        }
360    }
361
362    /// Unwrap the [`MeilisearchError`] from a [`Self::Failed`] [Task].
363    ///
364    /// Will panic if the task was not [`Self::Failed`].
365    ///
366    /// # Example
367    ///
368    /// ```
369    /// # use meilisearch_sdk::{client::*, indexes::*, errors::ErrorCode};
370    /// #
371    /// # let MEILISEARCH_URL = option_env!("MEILISEARCH_URL").unwrap_or("http://localhost:7700");
372    /// # let MEILISEARCH_API_KEY = option_env!("MEILISEARCH_API_KEY").unwrap_or("masterKey");
373    /// #
374    /// # tokio::runtime::Builder::new_current_thread().enable_all().build().unwrap().block_on(async {
375    /// # let client = Client::new(MEILISEARCH_URL, Some(MEILISEARCH_API_KEY)).unwrap();
376    /// let task = client.create_index("unwrap_failure", None).await.unwrap();
377    /// let task = client
378    ///     .create_index("unwrap_failure", None)
379    ///     .await
380    ///     .unwrap()
381    ///     .wait_for_completion(&client, None, None)
382    ///     .await
383    ///     .unwrap();
384    ///
385    /// assert!(task.is_failure());
386    ///
387    /// let failure = task.unwrap_failure();
388    ///
389    /// assert_eq!(failure.error_code, ErrorCode::IndexAlreadyExists);
390    /// # client.index("unwrap_failure").delete().await.unwrap().wait_for_completion(&client, None, None).await.unwrap();
391    /// # });
392    /// ```
393    #[must_use]
394    pub fn unwrap_failure(self) -> MeilisearchError {
395        match self {
396            Self::Failed {
397                content: FailedTask { error, .. },
398            } => error,
399            _ => panic!("Called `unwrap_failure` on a non `Failed` task."),
400        }
401    }
402
403    /// Returns `true` if the [Task] is [`Self::Failed`].
404    ///
405    /// # Example
406    ///
407    /// ```
408    /// # use meilisearch_sdk::{client::*, indexes::*, errors::ErrorCode};
409    /// #
410    /// # let MEILISEARCH_URL = option_env!("MEILISEARCH_URL").unwrap_or("http://localhost:7700");
411    /// # let MEILISEARCH_API_KEY = option_env!("MEILISEARCH_API_KEY").unwrap_or("masterKey");
412    /// #
413    /// # tokio::runtime::Builder::new_current_thread().enable_all().build().unwrap().block_on(async {
414    /// # let client = Client::new(MEILISEARCH_URL, Some(MEILISEARCH_API_KEY)).unwrap();
415    /// let task = client.create_index("is_failure", None).await.unwrap();
416    /// // create an index with a conflicting uid
417    /// let task = client
418    ///     .create_index("is_failure", None)
419    ///     .await
420    ///     .unwrap()
421    ///     .wait_for_completion(&client, None, None)
422    ///     .await
423    ///     .unwrap();
424    ///
425    /// assert!(task.is_failure());
426    /// # client.index("is_failure").delete().await.unwrap().wait_for_completion(&client, None, None).await.unwrap();
427    /// # });
428    /// ```
429    #[must_use]
430    pub fn is_failure(&self) -> bool {
431        matches!(self, Self::Failed { .. })
432    }
433
434    /// Returns `true` if the [Task] is [`Self::Succeeded`].
435    ///
436    /// # Example
437    ///
438    /// ```
439    /// # use meilisearch_sdk::{client::*, indexes::*, errors::ErrorCode};
440    /// #
441    /// # let MEILISEARCH_URL = option_env!("MEILISEARCH_URL").unwrap_or("http://localhost:7700");
442    /// # let MEILISEARCH_API_KEY = option_env!("MEILISEARCH_API_KEY").unwrap_or("masterKey");
443    /// #
444    /// # tokio::runtime::Builder::new_current_thread().enable_all().build().unwrap().block_on(async {
445    /// # let client = Client::new(MEILISEARCH_URL, Some(MEILISEARCH_API_KEY)).unwrap();
446    /// let task = client
447    ///     .create_index("is_success", None)
448    ///     .await
449    ///     .unwrap()
450    ///     .wait_for_completion(&client, None, None)
451    ///     .await
452    ///     .unwrap();
453    ///
454    /// assert!(task.is_success());
455    /// # task.try_make_index(&client).unwrap().delete().await.unwrap().wait_for_completion(&client, None, None).await.unwrap();
456    /// # });
457    /// ```
458    #[must_use]
459    pub fn is_success(&self) -> bool {
460        matches!(self, Self::Succeeded { .. })
461    }
462
463    /// Returns `true` if the [Task] is pending ([`Self::Enqueued`] or [`Self::Processing`]).
464    ///
465    /// # Example
466    /// ```no_run
467    /// # // The test is not run because it checks for an enqueued or processed status
468    /// # // and the task might already be processed when checking the status after the get_task call
469    /// # use meilisearch_sdk::{client::*, indexes::*, errors::ErrorCode};
470    /// #
471    /// # let MEILISEARCH_URL = option_env!("MEILISEARCH_URL").unwrap_or("http://localhost:7700");
472    /// # let MEILISEARCH_API_KEY = option_env!("MEILISEARCH_API_KEY").unwrap_or("masterKey");
473    /// #
474    /// # tokio::runtime::Builder::new_current_thread().enable_all().build().unwrap().block_on(async {
475    /// # let client = Client::new(MEILISEARCH_URL, Some(MEILISEARCH_API_KEY)).unwrap();
476    /// let task_info = client
477    ///     .create_index("is_pending", None)
478    ///     .await
479    ///     .unwrap();
480    /// let task = client.get_task(task_info).await.unwrap();
481    ///
482    /// assert!(task.is_pending());
483    /// # task.wait_for_completion(&client, None, None).await.unwrap().try_make_index(&client).unwrap().delete().await.unwrap().wait_for_completion(&client, None, None).await.unwrap();
484    /// # });
485    /// ```
486    #[must_use]
487    pub fn is_pending(&self) -> bool {
488        matches!(self, Self::Enqueued { .. } | Self::Processing { .. })
489    }
490}
491
492impl AsRef<u32> for Task {
493    fn as_ref(&self) -> &u32 {
494        match self {
495            Self::Enqueued { content } => content.as_ref(),
496            Self::Processing { content } => content.as_ref(),
497            Self::Succeeded { content } => content.as_ref(),
498            Self::Failed { content } => content.as_ref(),
499        }
500    }
501}
502
503#[derive(Debug, Serialize, Clone)]
504pub struct TasksPaginationFilters {
505    /// Maximum number of tasks to return.
506    #[serde(skip_serializing_if = "Option::is_none")]
507    limit: Option<u32>,
508    /// The first task uid that should be returned.
509    #[serde(skip_serializing_if = "Option::is_none")]
510    from: Option<u32>,
511}
512
513#[derive(Debug, Serialize, Clone)]
514pub struct TasksCancelFilters {}
515
516#[derive(Debug, Serialize, Clone)]
517pub struct TasksDeleteFilters {}
518
519pub type TasksSearchQuery<'a, Http> = TasksQuery<'a, TasksPaginationFilters, Http>;
520pub type TasksCancelQuery<'a, Http> = TasksQuery<'a, TasksCancelFilters, Http>;
521pub type TasksDeleteQuery<'a, Http> = TasksQuery<'a, TasksDeleteFilters, Http>;
522
523#[derive(Debug, Serialize, Clone)]
524#[serde(rename_all = "camelCase")]
525pub struct TasksQuery<'a, T, Http: HttpClient> {
526    #[serde(skip_serializing)]
527    client: &'a Client<Http>,
528    /// Index uids array to only retrieve the tasks of the indexes.
529    #[serde(skip_serializing_if = "Option::is_none")]
530    index_uids: Option<Vec<&'a str>>,
531    /// Statuses array to only retrieve the tasks with these statuses.
532    #[serde(skip_serializing_if = "Option::is_none")]
533    statuses: Option<Vec<&'a str>>,
534    /// Types array to only retrieve the tasks with these [`TaskType`]s.
535    #[serde(skip_serializing_if = "Option::is_none", rename = "types")]
536    task_types: Option<Vec<&'a str>>,
537    /// Uids of the tasks to retrieve.
538    #[serde(skip_serializing_if = "Option::is_none")]
539    uids: Option<Vec<&'a u32>>,
540    /// Uids of the tasks that canceled other tasks.
541    #[serde(skip_serializing_if = "Option::is_none")]
542    canceled_by: Option<Vec<&'a u32>>,
543    /// Date to retrieve all tasks that were enqueued before it.
544    #[serde(
545        skip_serializing_if = "Option::is_none",
546        serialize_with = "time::serde::rfc3339::option::serialize"
547    )]
548    before_enqueued_at: Option<OffsetDateTime>,
549    /// Date to retrieve all tasks that were enqueued after it.
550    #[serde(
551        skip_serializing_if = "Option::is_none",
552        serialize_with = "time::serde::rfc3339::option::serialize"
553    )]
554    after_enqueued_at: Option<OffsetDateTime>,
555    /// Date to retrieve all tasks that were started before it.
556    #[serde(
557        skip_serializing_if = "Option::is_none",
558        serialize_with = "time::serde::rfc3339::option::serialize"
559    )]
560    before_started_at: Option<OffsetDateTime>,
561    /// Date to retrieve all tasks that were started after it.
562    #[serde(
563        skip_serializing_if = "Option::is_none",
564        serialize_with = "time::serde::rfc3339::option::serialize"
565    )]
566    after_started_at: Option<OffsetDateTime>,
567    /// Date to retrieve all tasks that were finished before it.
568    #[serde(
569        skip_serializing_if = "Option::is_none",
570        serialize_with = "time::serde::rfc3339::option::serialize"
571    )]
572    before_finished_at: Option<OffsetDateTime>,
573    /// Date to retrieve all tasks that were finished after it.
574    #[serde(
575        skip_serializing_if = "Option::is_none",
576        serialize_with = "time::serde::rfc3339::option::serialize"
577    )]
578    after_finished_at: Option<OffsetDateTime>,
579
580    #[serde(flatten)]
581    pagination: T,
582
583    /// Whether to reverse the sort
584    #[serde(skip_serializing_if = "Option::is_none")]
585    reverse: Option<bool>,
586}
587
588#[allow(missing_docs)]
589impl<'a, T, Http: HttpClient> TasksQuery<'a, T, Http> {
590    pub fn with_index_uids<'b>(
591        &'b mut self,
592        index_uids: impl IntoIterator<Item = &'a str>,
593    ) -> &'b mut TasksQuery<'a, T, Http> {
594        self.index_uids = Some(index_uids.into_iter().collect());
595        self
596    }
597    pub fn with_statuses<'b>(
598        &'b mut self,
599        statuses: impl IntoIterator<Item = &'a str>,
600    ) -> &'b mut TasksQuery<'a, T, Http> {
601        self.statuses = Some(statuses.into_iter().collect());
602        self
603    }
604    pub fn with_types<'b>(
605        &'b mut self,
606        task_types: impl IntoIterator<Item = &'a str>,
607    ) -> &'b mut TasksQuery<'a, T, Http> {
608        self.task_types = Some(task_types.into_iter().collect());
609        self
610    }
611    pub fn with_uids<'b>(
612        &'b mut self,
613        uids: impl IntoIterator<Item = &'a u32>,
614    ) -> &'b mut TasksQuery<'a, T, Http> {
615        self.uids = Some(uids.into_iter().collect());
616        self
617    }
618    pub fn with_before_enqueued_at<'b>(
619        &'b mut self,
620        before_enqueued_at: &'a OffsetDateTime,
621    ) -> &'b mut TasksQuery<'a, T, Http> {
622        self.before_enqueued_at = Some(*before_enqueued_at);
623        self
624    }
625    pub fn with_after_enqueued_at<'b>(
626        &'b mut self,
627        after_enqueued_at: &'a OffsetDateTime,
628    ) -> &'b mut TasksQuery<'a, T, Http> {
629        self.after_enqueued_at = Some(*after_enqueued_at);
630        self
631    }
632    pub fn with_before_started_at<'b>(
633        &'b mut self,
634        before_started_at: &'a OffsetDateTime,
635    ) -> &'b mut TasksQuery<'a, T, Http> {
636        self.before_started_at = Some(*before_started_at);
637        self
638    }
639    pub fn with_after_started_at<'b>(
640        &'b mut self,
641        after_started_at: &'a OffsetDateTime,
642    ) -> &'b mut TasksQuery<'a, T, Http> {
643        self.after_started_at = Some(*after_started_at);
644        self
645    }
646    pub fn with_before_finished_at<'b>(
647        &'b mut self,
648        before_finished_at: &'a OffsetDateTime,
649    ) -> &'b mut TasksQuery<'a, T, Http> {
650        self.before_finished_at = Some(*before_finished_at);
651        self
652    }
653    pub fn with_after_finished_at<'b>(
654        &'b mut self,
655        after_finished_at: &'a OffsetDateTime,
656    ) -> &'b mut TasksQuery<'a, T, Http> {
657        self.after_finished_at = Some(*after_finished_at);
658        self
659    }
660    pub fn with_canceled_by<'b>(
661        &'b mut self,
662        task_uids: impl IntoIterator<Item = &'a u32>,
663    ) -> &'b mut TasksQuery<'a, T, Http> {
664        self.canceled_by = Some(task_uids.into_iter().collect());
665        self
666    }
667    pub fn with_reverse<'b>(&'b mut self, reverse: bool) -> &'b mut TasksQuery<'a, T, Http> {
668        self.reverse = Some(reverse);
669        self
670    }
671}
672
673impl<'a, Http: HttpClient> TasksQuery<'a, TasksCancelFilters, Http> {
674    #[must_use]
675    pub fn new(client: &'a Client<Http>) -> TasksQuery<'a, TasksCancelFilters, Http> {
676        TasksQuery {
677            client,
678            index_uids: None,
679            statuses: None,
680            task_types: None,
681            uids: None,
682            canceled_by: None,
683            before_enqueued_at: None,
684            after_enqueued_at: None,
685            before_started_at: None,
686            after_started_at: None,
687            before_finished_at: None,
688            after_finished_at: None,
689            reverse: None,
690            pagination: TasksCancelFilters {},
691        }
692    }
693
694    pub async fn execute(&'a self) -> Result<TaskInfo, Error> {
695        self.client.cancel_tasks_with(self).await
696    }
697}
698
699impl<'a, Http: HttpClient> TasksQuery<'a, TasksDeleteFilters, Http> {
700    #[must_use]
701    pub fn new(client: &'a Client<Http>) -> TasksQuery<'a, TasksDeleteFilters, Http> {
702        TasksQuery {
703            client,
704            index_uids: None,
705            statuses: None,
706            task_types: None,
707            uids: None,
708            canceled_by: None,
709            before_enqueued_at: None,
710            after_enqueued_at: None,
711            before_started_at: None,
712            after_started_at: None,
713            before_finished_at: None,
714            after_finished_at: None,
715            pagination: TasksDeleteFilters {},
716            reverse: None,
717        }
718    }
719
720    pub async fn execute(&'a self) -> Result<TaskInfo, Error> {
721        self.client.delete_tasks_with(self).await
722    }
723}
724
725impl<'a, Http: HttpClient> TasksQuery<'a, TasksPaginationFilters, Http> {
726    #[must_use]
727    pub fn new(client: &'a Client<Http>) -> TasksQuery<'a, TasksPaginationFilters, Http> {
728        TasksQuery {
729            client,
730            index_uids: None,
731            statuses: None,
732            task_types: None,
733            uids: None,
734            canceled_by: None,
735            before_enqueued_at: None,
736            after_enqueued_at: None,
737            before_started_at: None,
738            after_started_at: None,
739            before_finished_at: None,
740            after_finished_at: None,
741            pagination: TasksPaginationFilters {
742                limit: None,
743                from: None,
744            },
745            reverse: None,
746        }
747    }
748    pub fn with_limit<'b>(
749        &'b mut self,
750        limit: u32,
751    ) -> &'b mut TasksQuery<'a, TasksPaginationFilters, Http> {
752        self.pagination.limit = Some(limit);
753        self
754    }
755    pub fn with_from<'b>(
756        &'b mut self,
757        from: u32,
758    ) -> &'b mut TasksQuery<'a, TasksPaginationFilters, Http> {
759        self.pagination.from = Some(from);
760        self
761    }
762    pub async fn execute(&'a self) -> Result<TasksResults, Error> {
763        self.client.get_tasks_with(self).await
764    }
765}
766
767#[cfg(test)]
768mod test {
769
770    #[test]
771    fn test_deserialize_enqueued_task_with_remotes() {
772        let json = r#"{
773          "enqueuedAt": "2022-02-03T13:02:38.369634Z",
774          "indexUid": "movies",
775          "status": "enqueued",
776          "type": "indexUpdate",
777          "uid": 12,
778          "remotes": { "ms-00": { "status": "ok" } }
779        }"#;
780        let task: Task = serde_json::from_str(json).unwrap();
781        match task {
782            Task::Enqueued { content } => {
783                let remotes = content.remotes.expect("remotes should be present");
784                assert!(remotes.contains_key("ms-00"));
785            }
786            _ => panic!("expected enqueued task"),
787        }
788    }
789
790    #[test]
791    fn test_deserialize_processing_task_with_remotes() {
792        let json = r#"{
793          "details": {
794            "indexedDocuments": null,
795            "receivedDocuments": 10
796          },
797          "duration": null,
798          "enqueuedAt": "2022-02-03T15:17:02.801341Z",
799          "finishedAt": null,
800          "indexUid": "movies",
801          "startedAt": "2022-02-03T15:17:02.812338Z",
802          "status": "processing",
803          "type": "documentAdditionOrUpdate",
804          "uid": 14,
805          "remotes": { "ms-00": { "status": "ok" } }
806        }"#;
807        let task: Task = serde_json::from_str(json).unwrap();
808        match task {
809            Task::Processing { content } => {
810                let remotes = content.remotes.expect("remotes should be present");
811                assert!(remotes.contains_key("ms-00"));
812            }
813            _ => panic!("expected processing task"),
814        }
815    }
816
817    use super::*;
818    use crate::{
819        client::*,
820        errors::{ErrorCode, ErrorType},
821    };
822    use big_s::S;
823    use meilisearch_test_macro::meilisearch_test;
824    use serde::{Deserialize, Serialize};
825    use std::time::Duration;
826
827    #[derive(Debug, Serialize, Deserialize, PartialEq)]
828    struct Document {
829        id: usize,
830        value: String,
831        kind: String,
832    }
833
834    #[test]
835    fn test_deserialize_task() {
836        let datetime = OffsetDateTime::parse(
837            "2022-02-03T13:02:38.369634Z",
838            &time::format_description::well_known::Rfc3339,
839        )
840        .unwrap();
841
842        let task: Task = serde_json::from_str(
843            r#"
844{
845  "enqueuedAt": "2022-02-03T13:02:38.369634Z",
846  "indexUid": "meili",
847  "status": "enqueued",
848  "type": "documentAdditionOrUpdate",
849  "uid": 12
850}"#,
851        )
852        .unwrap();
853
854        assert!(matches!(
855            task,
856            Task::Enqueued {
857                content: EnqueuedTask {
858                    enqueued_at,
859                    index_uid: Some(index_uid),
860                    update_type: TaskType::DocumentAdditionOrUpdate { details: None },
861                    uid: 12, .. }
862            }
863        if enqueued_at == datetime && index_uid == "meili"));
864
865        let task: Task = serde_json::from_str(
866            r#"
867{
868  "details": {
869    "indexedDocuments": null,
870    "receivedDocuments": 19547
871  },
872  "duration": null,
873  "enqueuedAt": "2022-02-03T15:17:02.801341Z",
874  "finishedAt": null,
875  "indexUid": "meili",
876  "startedAt": "2022-02-03T15:17:02.812338Z",
877  "status": "processing",
878  "type": "documentAdditionOrUpdate",
879  "uid": 14
880}"#,
881        )
882        .unwrap();
883
884        assert!(matches!(
885            task,
886            Task::Processing {
887                content: ProcessingTask {
888                    started_at,
889                    update_type: TaskType::DocumentAdditionOrUpdate {
890                        details: Some(DocumentAdditionOrUpdate {
891                            received_documents: 19547,
892                            indexed_documents: None,
893                        })
894                    },
895                    uid: 14,
896                    ..
897                }
898            }
899            if started_at == OffsetDateTime::parse(
900                "2022-02-03T15:17:02.812338Z",
901                &time::format_description::well_known::Rfc3339
902            ).unwrap()
903        ));
904
905        let task: Task = serde_json::from_str(
906            r#"
907{
908  "details": {
909    "indexedDocuments": 19546,
910    "receivedDocuments": 19547
911  },
912  "duration": "PT10.848957S",
913  "enqueuedAt": "2022-02-03T15:17:02.801341Z",
914  "finishedAt": "2022-02-03T15:17:13.661295Z",
915  "indexUid": "meili",
916  "startedAt": "2022-02-03T15:17:02.812338Z",
917  "status": "succeeded",
918  "type": "documentAdditionOrUpdate",
919  "uid": 14
920}"#,
921        )
922        .unwrap();
923
924        assert!(matches!(
925            task,
926            Task::Succeeded {
927                content: SucceededTask {
928                    update_type: TaskType::DocumentAdditionOrUpdate {
929                        details: Some(DocumentAdditionOrUpdate {
930                            received_documents: 19547,
931                            indexed_documents: Some(19546),
932                        })
933                    },
934                    uid: 14,
935                    duration,
936                    ..
937                }
938            }
939            if duration == Duration::from_millis(10_848)
940        ));
941    }
942
943    #[meilisearch_test]
944    async fn test_wait_for_task_with_args(client: Client, movies: Index) -> Result<(), Error> {
945        let task = movies
946            .add_documents(
947                &[
948                    Document {
949                        id: 0,
950                        kind: "title".into(),
951                        value: S("The Social Network"),
952                    },
953                    Document {
954                        id: 1,
955                        kind: "title".into(),
956                        value: S("Harry Potter and the Sorcerer's Stone"),
957                    },
958                ],
959                None,
960            )
961            .await?
962            .wait_for_completion(
963                &client,
964                Some(Duration::from_millis(1)),
965                Some(Duration::from_millis(6000)),
966            )
967            .await?;
968
969        assert!(matches!(task, Task::Succeeded { .. }));
970        Ok(())
971    }
972
973    #[meilisearch_test]
974    async fn test_get_tasks_no_params() -> Result<(), Error> {
975        let mut s = mockito::Server::new_async().await;
976        let mock_server_url = s.url();
977        let client = Client::new(mock_server_url, Some("masterKey")).unwrap();
978        let path = "/tasks";
979
980        let mock_res = s.mock("GET", path).with_status(200).create_async().await;
981        let _ = client.get_tasks().await;
982        mock_res.assert_async().await;
983
984        Ok(())
985    }
986
987    #[meilisearch_test]
988    async fn test_get_tasks_with_params() -> Result<(), Error> {
989        let mut s = mockito::Server::new_async().await;
990        let mock_server_url = s.url();
991        let client = Client::new(mock_server_url, Some("masterKey")).unwrap();
992        let path =
993            "/tasks?indexUids=movies,test&statuses=equeued&types=documentDeletion&uids=1&limit=0&from=1&reverse=true";
994
995        let mock_res = s.mock("GET", path).with_status(200).create_async().await;
996
997        let mut query = TasksSearchQuery::new(&client);
998        query
999            .with_index_uids(["movies", "test"])
1000            .with_statuses(["equeued"])
1001            .with_types(["documentDeletion"])
1002            .with_from(1)
1003            .with_limit(0)
1004            .with_uids([&1])
1005            .with_reverse(true);
1006
1007        let _ = client.get_tasks_with(&query).await;
1008
1009        mock_res.assert_async().await;
1010
1011        Ok(())
1012    }
1013
1014    #[meilisearch_test]
1015    async fn test_get_tasks_with_date_params() -> Result<(), Error> {
1016        let mut s = mockito::Server::new_async().await;
1017        let mock_server_url = s.url();
1018        let client = Client::new(mock_server_url, Some("masterKey")).unwrap();
1019        let path = "/tasks?\
1020            beforeEnqueuedAt=2022-02-03T13%3A02%3A38.369634Z\
1021            &afterEnqueuedAt=2023-02-03T13%3A02%3A38.369634Z\
1022            &beforeStartedAt=2024-02-03T13%3A02%3A38.369634Z\
1023            &afterStartedAt=2025-02-03T13%3A02%3A38.369634Z\
1024            &beforeFinishedAt=2026-02-03T13%3A02%3A38.369634Z\
1025            &afterFinishedAt=2027-02-03T13%3A02%3A38.369634Z";
1026
1027        let mock_res = s.mock("GET", path).with_status(200).create_async().await;
1028
1029        let before_enqueued_at = OffsetDateTime::parse(
1030            "2022-02-03T13:02:38.369634Z",
1031            &time::format_description::well_known::Rfc3339,
1032        )
1033        .unwrap();
1034        let after_enqueued_at = OffsetDateTime::parse(
1035            "2023-02-03T13:02:38.369634Z",
1036            &time::format_description::well_known::Rfc3339,
1037        )
1038        .unwrap();
1039        let before_started_at = OffsetDateTime::parse(
1040            "2024-02-03T13:02:38.369634Z",
1041            &time::format_description::well_known::Rfc3339,
1042        )
1043        .unwrap();
1044
1045        let after_started_at = OffsetDateTime::parse(
1046            "2025-02-03T13:02:38.369634Z",
1047            &time::format_description::well_known::Rfc3339,
1048        )
1049        .unwrap();
1050
1051        let before_finished_at = OffsetDateTime::parse(
1052            "2026-02-03T13:02:38.369634Z",
1053            &time::format_description::well_known::Rfc3339,
1054        )
1055        .unwrap();
1056
1057        let after_finished_at = OffsetDateTime::parse(
1058            "2027-02-03T13:02:38.369634Z",
1059            &time::format_description::well_known::Rfc3339,
1060        )
1061        .unwrap();
1062
1063        let mut query = TasksSearchQuery::new(&client);
1064        query
1065            .with_before_enqueued_at(&before_enqueued_at)
1066            .with_after_enqueued_at(&after_enqueued_at)
1067            .with_before_started_at(&before_started_at)
1068            .with_after_started_at(&after_started_at)
1069            .with_before_finished_at(&before_finished_at)
1070            .with_after_finished_at(&after_finished_at);
1071
1072        let _ = client.get_tasks_with(&query).await;
1073
1074        mock_res.assert_async().await;
1075
1076        Ok(())
1077    }
1078
1079    #[meilisearch_test]
1080    async fn test_get_tasks_on_struct_with_params() -> Result<(), Error> {
1081        let mut s = mockito::Server::new_async().await;
1082        let mock_server_url = s.url();
1083        let client = Client::new(mock_server_url, Some("masterKey")).unwrap();
1084        let path =
1085            "/tasks?indexUids=movies,test&statuses=equeued&types=documentDeletion&canceledBy=9";
1086
1087        let mock_res = s.mock("GET", path).with_status(200).create_async().await;
1088
1089        let mut query = TasksSearchQuery::new(&client);
1090        let _ = query
1091            .with_index_uids(["movies", "test"])
1092            .with_statuses(["equeued"])
1093            .with_types(["documentDeletion"])
1094            .with_canceled_by([&9])
1095            .execute()
1096            .await;
1097
1098        mock_res.assert_async().await;
1099
1100        Ok(())
1101    }
1102
1103    #[meilisearch_test]
1104    async fn test_get_tasks_with_none_existant_index_uids(client: Client) -> Result<(), Error> {
1105        let mut query = TasksSearchQuery::new(&client);
1106        query.with_index_uids(["no_name"]);
1107        let tasks = client.get_tasks_with(&query).await.unwrap();
1108
1109        assert_eq!(tasks.results.len(), 0);
1110        Ok(())
1111    }
1112
1113    #[meilisearch_test]
1114    async fn test_get_tasks_with_execute(client: Client) -> Result<(), Error> {
1115        let tasks = TasksSearchQuery::new(&client)
1116            .with_index_uids(["no_name"])
1117            .execute()
1118            .await
1119            .unwrap();
1120
1121        assert_eq!(tasks.results.len(), 0);
1122        Ok(())
1123    }
1124
1125    #[meilisearch_test]
1126    async fn test_failing_task(client: Client, index: Index) -> Result<(), Error> {
1127        let task_info = client.create_index(index.uid, None).await.unwrap();
1128        let task = client.get_task(task_info).await?;
1129        let task = client.wait_for_task(task, None, None).await?;
1130
1131        let error = task.unwrap_failure();
1132        assert_eq!(error.error_code, ErrorCode::IndexAlreadyExists);
1133        assert_eq!(error.error_type, ErrorType::InvalidRequest);
1134        Ok(())
1135    }
1136
1137    #[meilisearch_test]
1138    async fn test_cancel_tasks_with_params() -> Result<(), Error> {
1139        let mut s = mockito::Server::new_async().await;
1140        let mock_server_url = s.url();
1141        let client = Client::new(mock_server_url, Some("masterKey")).unwrap();
1142        let path =
1143            "/tasks/cancel?indexUids=movies,test&statuses=equeued&types=documentDeletion&uids=1";
1144
1145        let mock_res = s.mock("POST", path).with_status(200).create_async().await;
1146
1147        let mut query = TasksCancelQuery::new(&client);
1148        query
1149            .with_index_uids(["movies", "test"])
1150            .with_statuses(["equeued"])
1151            .with_types(["documentDeletion"])
1152            .with_uids([&1]);
1153
1154        let _ = client.cancel_tasks_with(&query).await;
1155
1156        mock_res.assert_async().await;
1157
1158        Ok(())
1159    }
1160
1161    #[meilisearch_test]
1162    async fn test_cancel_tasks_with_params_execute() -> Result<(), Error> {
1163        let mut s = mockito::Server::new_async().await;
1164        let mock_server_url = s.url();
1165        let client = Client::new(mock_server_url, Some("masterKey")).unwrap();
1166        let path =
1167            "/tasks/cancel?indexUids=movies,test&statuses=equeued&types=documentDeletion&uids=1";
1168
1169        let mock_res = s.mock("POST", path).with_status(200).create_async().await;
1170
1171        let mut query = TasksCancelQuery::new(&client);
1172        let _ = query
1173            .with_index_uids(["movies", "test"])
1174            .with_statuses(["equeued"])
1175            .with_types(["documentDeletion"])
1176            .with_uids([&1])
1177            .execute()
1178            .await;
1179
1180        mock_res.assert_async().await;
1181
1182        Ok(())
1183    }
1184
1185    #[meilisearch_test]
1186    async fn test_delete_tasks_with_params() -> Result<(), Error> {
1187        let mut s = mockito::Server::new_async().await;
1188        let mock_server_url = s.url();
1189        let client = Client::new(mock_server_url, Some("masterKey")).unwrap();
1190        let path = "/tasks?indexUids=movies,test&statuses=equeued&types=documentDeletion&uids=1";
1191
1192        let mock_res = s.mock("DELETE", path).with_status(200).create_async().await;
1193
1194        let mut query = TasksDeleteQuery::new(&client);
1195        query
1196            .with_index_uids(["movies", "test"])
1197            .with_statuses(["equeued"])
1198            .with_types(["documentDeletion"])
1199            .with_uids([&1]);
1200
1201        let _ = client.delete_tasks_with(&query).await;
1202
1203        mock_res.assert_async().await;
1204
1205        Ok(())
1206    }
1207
1208    #[meilisearch_test]
1209    async fn test_delete_tasks_with_params_execute() -> Result<(), Error> {
1210        let mut s = mockito::Server::new_async().await;
1211        let mock_server_url = s.url();
1212        let client = Client::new(mock_server_url, Some("masterKey")).unwrap();
1213        let path = "/tasks?indexUids=movies,test&statuses=equeued&types=documentDeletion&uids=1";
1214
1215        let mock_res = s.mock("DELETE", path).with_status(200).create_async().await;
1216
1217        let mut query = TasksDeleteQuery::new(&client);
1218        let _ = query
1219            .with_index_uids(["movies", "test"])
1220            .with_statuses(["equeued"])
1221            .with_types(["documentDeletion"])
1222            .with_uids([&1])
1223            .execute()
1224            .await;
1225
1226        mock_res.assert_async().await;
1227
1228        Ok(())
1229    }
1230}
meilisearch_sdk/tasks.rs

meilisearch_sdk/
tasks.rs
