cognite/api/
resource.rs

1use std::collections::VecDeque;
2use std::future::Future;
3use std::{marker::PhantomData, sync::Arc};
4
5use futures::future::try_join_all;
6use futures::stream::{try_unfold, SelectAll};
7use futures::TryStream;
8use serde::{de::DeserializeOwned, Serialize};
9
10use crate::dto::items::*;
11use crate::{
12    ApiClient, CondBoxedStream, CondSend, EqIdentity, Filter, Identity, IntoParams, IntoPatch,
13    Partition, Patch, Result, Search, SetCursor, UpsertOptions, WithPartition,
14};
15
16use super::utils::{get_duplicates_from_result, get_missing_from_result};
17
18/// A resource instance contains methods for accessing a single
19/// CDF resource type.
20pub struct Resource<T> {
21    /// A reference to the shared API Client.
22    pub api_client: Arc<ApiClient>,
23    marker: PhantomData<T>,
24}
25
26impl<T> Resource<T> {
27    /// Create a new resource with given API client.
28    ///
29    /// # Arguments
30    ///
31    /// * `api_client` - API client reference.
32    pub fn new(api_client: Arc<ApiClient>) -> Self {
33        Resource {
34            api_client,
35            marker: PhantomData,
36        }
37    }
38}
39
40impl<T> Clone for Resource<T> {
41    fn clone(&self) -> Self {
42        Self {
43            api_client: self.api_client.clone(),
44            marker: PhantomData,
45        }
46    }
47}
48
49impl<T> WithApiClient for Resource<T> {
50    fn get_client(&self) -> &ApiClient {
51        &self.api_client
52    }
53}
54
55/// Trait for a type that contains an API client.
56pub trait WithApiClient {
57    /// Get the API client for this type.
58    fn get_client(&self) -> &ApiClient;
59}
60
61/// Trait for a type with a base path.
62pub trait WithBasePath {
63    /// Base path for this resource type.
64    const BASE_PATH: &'static str;
65}
66
67/// Trait for simple GET / endpoints.
68pub trait List<TParams, TResponse>
69where
70    TParams: IntoParams + Send + Sync + 'static,
71    TResponse: Serialize + DeserializeOwned + Send + Sync,
72    Self: WithApiClient + WithBasePath + Sync,
73{
74    /// Query a resource with optional query parameters.
75    ///
76    /// # Arguments
77    ///
78    /// * `params` - Query parameters.
79    fn list(
80        &self,
81        params: Option<TParams>,
82    ) -> impl Future<Output = Result<ItemsVec<TResponse, Cursor>>> + CondSend {
83        async move {
84            self.get_client()
85                .get_with_params(Self::BASE_PATH, params)
86                .await
87        }
88    }
89
90    /// Query a resource with query parameters, continuing until the cursor is exhausted.
91    ///
92    /// # Arguments
93    ///
94    /// * `params` - Initial query parameters. This can contain a cursor which is the starting point.
95    fn list_all(
96        &self,
97        mut params: TParams,
98    ) -> impl Future<Output = Result<Vec<TResponse>>> + CondSend
99    where
100        TParams: SetCursor + Clone,
101        TResponse: Send,
102    {
103        async move {
104            let mut result = vec![];
105            loop {
106                let lparams = params.clone();
107                let response: ItemsVec<TResponse, Cursor> = self
108                    .get_client()
109                    .get_with_params(Self::BASE_PATH, Some(lparams))
110                    .await?;
111                for it in response.items {
112                    result.push(it);
113                }
114                match response.extra_fields.next_cursor {
115                    Some(cursor) => params.set_cursor(Some(cursor)),
116                    None => return Ok(result),
117                }
118            }
119        }
120    }
121
122    /// List resources, following cursors. This returns a stream, you can abort the stream whenever you
123    /// want and only resources retrieved up to that point will be returned.
124    ///
125    /// Each item in the stream will be a result, after the first error is returned the
126    /// stream will end.
127    ///
128    /// # Arguments
129    ///
130    /// * `params` - Initial query parameters. This can contain a cursors used as starting point.
131    fn list_all_stream(
132        &self,
133        params: TParams,
134    ) -> impl TryStream<Ok = TResponse, Error = crate::Error, Item = Result<TResponse>> + CondSend
135    where
136        TParams: SetCursor + Clone,
137        TResponse: Send + 'static,
138    {
139        let state = CursorStreamState {
140            req: params,
141            responses: VecDeque::new(),
142            next_cursor: CursorState::Initial,
143        };
144
145        try_unfold(state, move |mut state| async move {
146            if let Some(next) = state.responses.pop_front() {
147                Ok(Some((next, state)))
148            } else {
149                let cursor = match std::mem::take(&mut state.next_cursor) {
150                    CursorState::Initial => None,
151                    CursorState::Some(x) => Some(x),
152                    CursorState::End => {
153                        return Ok(None);
154                    }
155                };
156                state.req.set_cursor(cursor);
157                let response: ItemsVec<TResponse, Cursor> = self
158                    .get_client()
159                    .get_with_params(Self::BASE_PATH, Some(state.req.clone()))
160                    .await?;
161
162                state.responses.extend(response.items);
163                state.next_cursor = match response.extra_fields.next_cursor {
164                    Some(x) => CursorState::Some(x),
165                    None => CursorState::End,
166                };
167                if let Some(next) = state.responses.pop_front() {
168                    Ok(Some((next, state)))
169                } else {
170                    Ok(None)
171                }
172            }
173        })
174    }
175}
176
177/// Trait for creating resources with POST / requests.
178pub trait Create<TCreate, TResponse>
179where
180    TCreate: Serialize + Sync + Send,
181    TResponse: Serialize + DeserializeOwned + Send,
182    Self: WithApiClient + WithBasePath + Sync,
183{
184    /// Create a list of resources.
185    ///
186    /// # Arguments
187    ///
188    /// `creates` - List of resources to create.
189    fn create(
190        &self,
191        creates: &[TCreate],
192    ) -> impl Future<Output = Result<Vec<TResponse>>> + CondSend {
193        async move {
194            let items = Items::new(creates);
195            let response: ItemsVec<TResponse> =
196                self.get_client().post(Self::BASE_PATH, &items).await?;
197            Ok(response.items)
198        }
199    }
200
201    /// Create a list of resources, converting from a different type.
202    ///
203    /// # Arguments
204    ///
205    /// * `creates` - List of resources to create.
206    fn create_from(
207        &self,
208        creates: &[impl Into<TCreate> + Sync + Clone],
209    ) -> impl Future<Output = Result<Vec<TResponse>>> + CondSend {
210        async move {
211            let to_add: Vec<TCreate> = creates.iter().map(|i| i.clone().into()).collect();
212            self.create(&to_add).await
213        }
214    }
215
216    /// Create a list of resources, ignoring any that fail with general "conflict" errors.
217    ///
218    /// # Arguments
219    ///
220    /// * `creates` - List of resources to create.
221    fn create_ignore_duplicates(
222        &self,
223        creates: &[TCreate],
224    ) -> impl Future<Output = Result<Vec<TResponse>>> + CondSend
225    where
226        TCreate: EqIdentity,
227    {
228        async move {
229            let resp = self.create(creates).await;
230
231            let duplicates: Option<Vec<Identity>> = get_duplicates_from_result(&resp);
232
233            if let Some(duplicates) = duplicates {
234                let next: Vec<&TCreate> = creates
235                    .iter()
236                    .filter(|c| !duplicates.iter().any(|i| c.eq(i)))
237                    .collect();
238
239                if next.is_empty() {
240                    if duplicates.len() == creates.len() {
241                        return Ok(vec![]);
242                    }
243                    return resp;
244                }
245
246                let items = Items::new(next);
247                let response: ItemsVec<TResponse> =
248                    self.get_client().post(Self::BASE_PATH, &items).await?;
249                Ok(response.items)
250            } else {
251                resp
252            }
253        }
254    }
255
256    /// Create a list of resources, converting from a different type, and ignoring any that fail
257    /// with general "conflict" errors.
258    ///
259    /// # Arguments
260    ///
261    /// * `creates` - List of resources to create.
262    fn create_from_ignore_duplicates<'a, T: 'a>(
263        &self,
264        creates: &'a [impl Into<TCreate> + Sync + Clone],
265    ) -> impl Future<Output = Result<Vec<TResponse>>> + CondSend
266    where
267        TCreate: EqIdentity,
268    {
269        async move {
270            let to_add: Vec<TCreate> = creates.iter().map(|i| i.clone().into()).collect();
271            self.create_ignore_duplicates(&to_add).await
272        }
273    }
274}
275
276/// Trait for upserts of resources that support both Create and Update.
277pub trait Upsert<'a, TCreate, TUpdate, TResponse>
278where
279    TCreate: Serialize + Sync + Send + EqIdentity + 'a + Clone + IntoPatch<TUpdate>,
280    TUpdate: Serialize + Sync + Send + Default,
281    TResponse: Serialize + DeserializeOwned + Sync + Send,
282    Self: WithApiClient + WithBasePath + Sync,
283{
284    /// Upsert a list resources, meaning that they will first be attempted created,
285    /// and if that fails with a conflict, update any that already existed, and create
286    /// the remainder.
287    ///
288    /// # Arguments
289    ///
290    /// * `upserts` - Resources to insert or update.
291    /// * `options` - Configuration for upserts, which fields are kept and which are overwritten.
292    fn upsert(
293        &'a self,
294        upserts: &'a [TCreate],
295        options: &UpsertOptions,
296    ) -> impl Future<Output = Result<Vec<TResponse>>> + CondSend {
297        async move {
298            let items = Items::new(upserts);
299            let resp: Result<ItemsVec<TResponse>> =
300                self.get_client().post(Self::BASE_PATH, &items).await;
301
302            let duplicates: Option<Vec<Identity>> = get_duplicates_from_result(&resp);
303
304            if let Some(duplicates) = duplicates {
305                let mut to_create = Vec::with_capacity(upserts.len() - duplicates.len());
306                let mut to_update = Vec::with_capacity(duplicates.len());
307                for it in upserts {
308                    let idt = duplicates.iter().find(|i| it.eq(i));
309                    if let Some(idt) = idt {
310                        to_update.push(Patch::<TUpdate> {
311                            id: idt.clone(),
312                            update: it.clone().patch(options),
313                        });
314                    } else {
315                        to_create.push(it);
316                    }
317                }
318
319                let mut result = Vec::with_capacity(to_create.len() + to_update.len());
320                if !to_create.is_empty() {
321                    let mut create_response: ItemsVec<TResponse> = self
322                        .get_client()
323                        .post(Self::BASE_PATH, &Items::new(to_create))
324                        .await?;
325                    result.append(&mut create_response.items);
326                }
327                if !to_update.is_empty() {
328                    let mut update_response: ItemsVec<TResponse> = self
329                        .get_client()
330                        .post(
331                            &format!("{}/update", Self::BASE_PATH),
332                            &Items::new(&to_update),
333                        )
334                        .await?;
335                    result.append(&mut update_response.items);
336                }
337
338                Ok(result)
339            } else {
340                resp.map(|i| i.items)
341            }
342        }
343    }
344}
345
346impl<'a, T, TCreate, TUpdate, TResponse> Upsert<'a, TCreate, TUpdate, TResponse> for T
347where
348    T: Create<TCreate, TResponse> + Update<Patch<TUpdate>, TResponse> + Sync,
349    TCreate: Serialize + Sync + Send + EqIdentity + 'a + Clone + IntoPatch<TUpdate>,
350    TUpdate: Serialize + Sync + Send + Default,
351    TResponse: Serialize + DeserializeOwned + Sync + Send,
352{
353}
354
355/// Trait for resource types that support upserts directly.
356pub trait UpsertCollection<TUpsert, TResponse> {
357    /// Upsert a list of resources.
358    ///
359    /// # Arguments
360    ///
361    /// * `collection` - Items to insert or update.
362    fn upsert(
363        &self,
364        collection: &TUpsert,
365    ) -> impl Future<Output = Result<Vec<TResponse>>> + CondSend
366    where
367        TUpsert: Serialize + Sync + Send,
368        TResponse: Serialize + DeserializeOwned + Sync + Send,
369        Self: WithApiClient + WithBasePath + Sync,
370    {
371        async move {
372            let response: ItemsVec<TResponse> =
373                self.get_client().post(Self::BASE_PATH, &collection).await?;
374            Ok(response.items)
375        }
376    }
377}
378
379/// Trait for resource types that can be deleted with a list of `TIdt`.
380pub trait Delete<TIdt>
381where
382    TIdt: Serialize + Sync + Send,
383    Self: WithApiClient + WithBasePath + Sync,
384{
385    /// Delete a list of resources by ID.
386    ///
387    /// # Arguments
388    ///
389    /// * `deletes` - IDs of items to delete.
390    fn delete(&self, deletes: &[TIdt]) -> impl Future<Output = Result<()>> + CondSend {
391        async move {
392            let items = Items::new(deletes);
393            self.get_client()
394                .post::<::serde_json::Value, Items<&[TIdt]>>(
395                    &format!("{}/delete", Self::BASE_PATH),
396                    &items,
397                )
398                .await?;
399            Ok(())
400        }
401    }
402}
403
404/// Trait for resource types that can be deleted with a more complex request.
405pub trait DeleteWithRequest<TReq>
406where
407    TReq: Serialize + Sync + Send,
408    Self: WithApiClient + WithBasePath + Sync,
409{
410    /// Delete resources using `req`.
411    ///
412    /// # Arguments
413    ///
414    /// * `req` - Request describing items to delete.
415    fn delete(&self, req: &TReq) -> impl Future<Output = Result<()>> + CondSend {
416        async move {
417            self.get_client()
418                .post::<::serde_json::Value, TReq>(&format!("{}/delete", Self::BASE_PATH), req)
419                .await?;
420            Ok(())
421        }
422    }
423}
424
425/// Trait for resource types that can be deleted with a list of identities and
426/// a boolean option to ignore unknown ids.
427pub trait DeleteWithIgnoreUnknownIds<TIdt>
428where
429    TIdt: Serialize + Sync + Send,
430    Self: WithApiClient + WithBasePath + Sync,
431{
432    /// Delete a list of resources, optionally ignore unknown ids.
433    ///
434    /// # Arguments
435    ///
436    /// * `deletes` - IDs of items to delete.
437    /// * `ignore_unknown_ids` - If `true`, missing IDs will be ignored, and not
438    ///   cause the request to fail.
439    fn delete(
440        &self,
441        deletes: impl Into<TIdt> + Send,
442        ignore_unknown_ids: bool,
443    ) -> impl Future<Output = Result<()>> + CondSend
444    where
445        Self: Sync,
446    {
447        async move {
448            let req = Items::new_with_extra_fields(
449                deletes.into(),
450                IgnoreUnknownIds { ignore_unknown_ids },
451            );
452            self.get_client()
453                .post::<::serde_json::Value, _>(&format!("{}/delete", Self::BASE_PATH), &req)
454                .await?;
455            Ok(())
456        }
457    }
458}
459
460/// Trait for resource types that can be deleted, and where the delete request
461/// has a non-empty response.
462pub trait DeleteWithResponse<TIdt, TResponse>
463where
464    TIdt: Serialize + Sync + Send,
465    TResponse: Serialize + DeserializeOwned + Sync + Send,
466    Self: WithApiClient + WithBasePath + Sync,
467{
468    /// Delete a list of resources.
469    ///
470    /// # Arguments
471    ///
472    /// * `deletes` - IDs of items to delete.
473    fn delete(
474        &self,
475        deletes: &[TIdt],
476    ) -> impl Future<Output = Result<ItemsVec<TResponse>>> + CondSend {
477        async move {
478            let items = Items::new(deletes);
479            let response: ItemsVec<TResponse> = self
480                .get_client()
481                .post(&format!("{}/delete", Self::BASE_PATH), &items)
482                .await?;
483            Ok(response)
484        }
485    }
486}
487
488/// Trait for resource types that can be patch updated.
489pub trait Update<TUpdate, TResponse>
490where
491    TUpdate: Serialize + Sync + Send,
492    TResponse: Serialize + DeserializeOwned,
493    Self: WithApiClient + WithBasePath + Sync,
494{
495    /// Update a list of resources.
496    ///
497    /// # Arguments
498    ///
499    /// * `updates` - Items to update.
500    fn update(
501        &self,
502        updates: &[TUpdate],
503    ) -> impl Future<Output = Result<Vec<TResponse>>> + CondSend {
504        async move {
505            let items = Items::new(updates);
506            let response: ItemsVec<TResponse> = self
507                .get_client()
508                .post(&format!("{}/update", Self::BASE_PATH), &items)
509                .await?;
510            Ok(response.items)
511        }
512    }
513
514    /// Update a list of resources by converting to the update from a different type.
515    ///
516    /// # Arguments
517    ///
518    /// * `updates` - Items to update.
519    fn update_from<'a, T>(
520        &self,
521        updates: &'a [T],
522    ) -> impl Future<Output = Result<Vec<TResponse>>> + CondSend
523    where
524        T: std::marker::Sync + Clone + 'a,
525        TUpdate: From<T>,
526    {
527        async move {
528            let to_update: Vec<TUpdate> =
529                updates.iter().map(|i| TUpdate::from(i.clone())).collect();
530            self.update(&to_update).await
531        }
532    }
533
534    /// Update a list of resources, ignoring any that fail due to items missing in CDF.
535    ///
536    /// # Arguments
537    ///
538    /// * `updates` - Items to update.
539    fn update_ignore_unknown_ids(
540        &self,
541        updates: &[TUpdate],
542    ) -> impl Future<Output = Result<Vec<TResponse>>> + CondSend
543    where
544        TUpdate: EqIdentity,
545        TResponse: Send,
546    {
547        async move {
548            let response = self.update(updates).await;
549            let missing: Option<Vec<Identity>> = get_missing_from_result(&response);
550
551            if let Some(missing) = missing {
552                let next: Vec<&TUpdate> = updates
553                    .iter()
554                    .filter(|c| !missing.iter().any(|i| c.eq(i)))
555                    .collect();
556
557                if next.is_empty() {
558                    if missing.len() == updates.len() {
559                        return Ok(vec![]);
560                    }
561                    return response;
562                }
563
564                let items = Items::new(next);
565                let response: ItemsVec<TResponse> = self
566                    .get_client()
567                    .post(&format!("{}/update", Self::BASE_PATH), &items)
568                    .await?;
569                Ok(response.items)
570            } else {
571                response
572            }
573        }
574    }
575
576    /// Update a list of resources by converting from a different type, ignoring any that fail
577    /// due items missing in CDF.
578    ///
579    /// # Arguments
580    ///
581    /// * `updates` - Items to update.
582    fn update_from_ignore_unknown_ids<'a, T>(
583        &self,
584        updates: &'a [T],
585    ) -> impl Future<Output = Result<Vec<TResponse>>> + CondSend
586    where
587        T: Sync + Clone + 'a,
588        TUpdate: From<T> + EqIdentity,
589        TResponse: Send,
590    {
591        async move {
592            let to_update: Vec<TUpdate> =
593                updates.iter().map(|i| TUpdate::from(i.clone())).collect();
594            self.update_ignore_unknown_ids(&to_update).await
595        }
596    }
597}
598
599/// Trait for retrieving items from CDF by id.
600pub trait Retrieve<TIdt, TResponse>
601where
602    TIdt: Serialize + Sync + Send,
603    TResponse: Serialize + DeserializeOwned,
604    Self: WithApiClient + WithBasePath + Sync,
605{
606    /// Retrieve a list of items from CDF by id.
607    ///
608    /// # Arguments
609    ///
610    /// * `ids` - IDs of items to retrieve.
611    fn retrieve(&self, ids: &[TIdt]) -> impl Future<Output = Result<Vec<TResponse>>> + CondSend {
612        async move {
613            let items = Items::new(ids);
614            let response: ItemsVec<TResponse> = self
615                .get_client()
616                .post(&format!("{}/byids", Self::BASE_PATH), &items)
617                .await?;
618            Ok(response.items)
619        }
620    }
621}
622
623/// Trait for retrieving items from CDF with a more complex request type.
624pub trait RetrieveWithRequest<TRequest, TResponse>
625where
626    TRequest: Serialize + Sync + Send,
627    TResponse: Serialize + DeserializeOwned,
628    Self: WithApiClient + WithBasePath + Sync,
629{
630    /// Retrieve items from CDF with a more complex request.
631    ///
632    /// # Arguments
633    ///
634    /// * `req` - Request describing items to retrieve.
635    fn retrieve(&self, req: &TRequest) -> impl Future<Output = Result<TResponse>> + CondSend {
636        async move {
637            let response: TResponse = self
638                .get_client()
639                .post(&format!("{}/byids", Self::BASE_PATH), req)
640                .await?;
641            Ok(response)
642        }
643    }
644}
645
646/// Trait for retrieving items from CDF with an option to ignore unknown IDs.
647pub trait RetrieveWithIgnoreUnknownIds<TIdt, TResponse>
648where
649    TIdt: Serialize + Sync + Send,
650    TResponse: Serialize + DeserializeOwned,
651    Self: WithApiClient + WithBasePath + Sync,
652{
653    /// Retrieve a list of items from CDF. If ignore_unknown_ids is false,
654    /// this will fail if any items are missing from CDF.
655    ///
656    /// # Arguments
657    ///
658    /// * `ids` - IDs of items to retrieve.
659    /// * `ignore_unknown_ids` - If `true`, items missing from CDF will be ignored, and not
660    ///   cause the request to fail.
661    fn retrieve(
662        &self,
663        ids: impl Into<TIdt> + Send,
664        ignore_unknown_ids: bool,
665    ) -> impl Future<Output = Result<Vec<TResponse>>> + CondSend {
666        async move {
667            let items =
668                Items::new_with_extra_fields(ids.into(), IgnoreUnknownIds { ignore_unknown_ids });
669            let response: ItemsVec<TResponse> = self
670                .get_client()
671                .post(&format!("{}/byids", Self::BASE_PATH), &items)
672                .await?;
673            Ok(response.items)
674        }
675    }
676}
677
678/// Trait for resource types that allow filtering with a simple filter.
679pub trait FilterItems<TFilter, TResponse>
680where
681    TFilter: Serialize + Sync + Send + 'static,
682    TResponse: Serialize + DeserializeOwned,
683    Self: WithApiClient + WithBasePath + Sync,
684{
685    /// Filter resources using a simple filter.
686    /// The response may contain a cursor that can be used to paginate results.
687    ///
688    /// # Arguments
689    ///
690    /// * `filter` - Filter which items to retrieve.
691    /// * `cursor` - Optional cursor for pagination.
692    /// * `limit` - Maximum number of result items.
693    fn filter_items(
694        &self,
695        filter: TFilter,
696        cursor: Option<String>,
697        limit: Option<u32>,
698    ) -> impl Future<Output = Result<ItemsVec<TResponse, Cursor>>> + CondSend {
699        async move {
700            let filter = Filter::<TFilter>::new(filter, cursor, limit);
701            let response: ItemsVec<TResponse, Cursor> = self
702                .get_client()
703                .post(&format!("{}/list", Self::BASE_PATH), &filter)
704                .await?;
705            Ok(response)
706        }
707    }
708}
709
710impl<TFilter, TResponse, T> FilterWithRequest<Filter<TFilter>, TResponse> for T
711where
712    TFilter: Serialize + Sync + Send + 'static,
713    TResponse: Serialize + DeserializeOwned,
714    T: FilterItems<TFilter, TResponse>,
715    Self: WithApiClient + WithBasePath,
716{
717}
718
719#[derive(Debug, Default)]
720pub(crate) enum CursorState {
721    Initial,
722    Some(String),
723    #[default]
724    End,
725}
726
727pub(crate) struct CursorStreamState<TFilter, TResponse> {
728    pub(crate) req: TFilter,
729    pub(crate) responses: VecDeque<TResponse>,
730    pub(crate) next_cursor: CursorState,
731}
732
733/// Trait for resource types that allow filtering with a more complex request.
734pub trait FilterWithRequest<TFilter, TResponse>
735where
736    TFilter: Serialize + Sync + Send + 'static,
737    TResponse: Serialize + DeserializeOwned,
738    Self: WithApiClient + WithBasePath + Sync,
739{
740    /// Filter resources.
741    ///
742    /// # Arguments
743    ///
744    /// * `filter` - Filter which items to retrieve.
745    fn filter(
746        &self,
747        filter: TFilter,
748    ) -> impl Future<Output = Result<ItemsVec<TResponse, Cursor>>> + CondSend {
749        async move {
750            let response: ItemsVec<TResponse, Cursor> = self
751                .get_client()
752                .post(&format!("{}/list", Self::BASE_PATH), &filter)
753                .await?;
754            Ok(response)
755        }
756    }
757
758    /// Filter resources, following cursors until they are exhausted.
759    ///
760    /// # Arguments
761    ///
762    /// * `filter` - Filter which items to retrieve.
763    fn filter_all(
764        &self,
765        mut filter: TFilter,
766    ) -> impl Future<Output = Result<Vec<TResponse>>> + CondSend
767    where
768        TFilter: SetCursor,
769        TResponse: Send,
770    {
771        async move {
772            let mut result = vec![];
773            loop {
774                let response: ItemsVec<TResponse, Cursor> = self
775                    .get_client()
776                    .post(&format!("{}/list", Self::BASE_PATH), &filter)
777                    .await?;
778                for it in response.items {
779                    result.push(it);
780                }
781                match response.extra_fields.next_cursor {
782                    Some(cursor) => filter.set_cursor(Some(cursor)),
783                    None => return Ok(result),
784                }
785            }
786        }
787    }
788
789    /// Filter resources, following cursors. This returns a stream, you can abort the stream whenever you
790    /// want and only resources retrieved up to that point will be returned.
791    ///
792    /// Each item in the stream will be a result, after the first error is returned the
793    /// stream will end.
794    ///
795    /// # Arguments
796    ///
797    /// * `filter` - Filter which items to retrieve.
798    fn filter_all_stream(
799        &self,
800        filter: TFilter,
801    ) -> impl TryStream<Ok = TResponse, Error = crate::Error, Item = Result<TResponse>> + CondSend
802    where
803        TFilter: SetCursor,
804        TResponse: Send + 'static,
805    {
806        let state = CursorStreamState {
807            req: filter,
808            responses: VecDeque::new(),
809            next_cursor: CursorState::Initial,
810        };
811
812        try_unfold(state, move |mut state| async move {
813            if let Some(next) = state.responses.pop_front() {
814                Ok(Some((next, state)))
815            } else {
816                let cursor = match std::mem::take(&mut state.next_cursor) {
817                    CursorState::Initial => None,
818                    CursorState::Some(x) => Some(x),
819                    CursorState::End => {
820                        return Ok(None);
821                    }
822                };
823                state.req.set_cursor(cursor);
824                let response: ItemsVec<TResponse, Cursor> = self
825                    .get_client()
826                    .post(&format!("{}/list", Self::BASE_PATH), &state.req)
827                    .await?;
828
829                state.responses.extend(response.items);
830                state.next_cursor = match response.extra_fields.next_cursor {
831                    Some(x) => CursorState::Some(x),
832                    None => CursorState::End,
833                };
834                if let Some(next) = state.responses.pop_front() {
835                    Ok(Some((next, state)))
836                } else {
837                    Ok(None)
838                }
839            }
840        })
841    }
842
843    /// Filter resources using partitioned reads, following cursors until all partitions are
844    /// exhausted.
845    ///
846    /// # Arguments
847    ///
848    /// * `filter` - Filter which items to retrieve.
849    /// * `num_partitions` - Number of partitions.
850    fn filter_all_partitioned(
851        &self,
852        filter: TFilter,
853        num_partitions: u32,
854    ) -> impl Future<Output = Result<Vec<TResponse>>> + CondSend
855    where
856        TFilter: SetCursor + WithPartition,
857        TResponse: Send,
858    {
859        async move {
860            let mut futures = Vec::with_capacity(num_partitions as usize);
861            for partition in 0..num_partitions {
862                let part_filter =
863                    filter.with_partition(Partition::new(partition + 1, num_partitions));
864                futures.push(self.filter_all(part_filter));
865            }
866            let results = try_join_all(futures).await?;
867            let mut response_items = Vec::with_capacity(results.iter().map(|i| i.len()).sum());
868            for chunk in results.into_iter() {
869                response_items.extend(chunk);
870            }
871            Ok(response_items)
872        }
873    }
874
875    /// Filter resources using partitioned reads, following cursors until all partitions
876    /// are exhausted. This returns a stream.
877    ///
878    /// Note that the returned stream is simply a combinator of streams returned by
879    /// `filter_all_stream` for different partitions.
880    ///
881    /// The order of the returned values is not guaranteed to be in any way consistent.
882    ///
883    /// # Arguments
884    ///
885    /// * `filter` - Filter which items to retrieve.
886    /// * `num_partitions` - Number of partitions.
887    fn filter_all_partitioned_stream(
888        &self,
889        filter: TFilter,
890        num_partitions: u32,
891    ) -> impl TryStream<Ok = TResponse, Error = crate::Error, Item = Result<TResponse>> + CondSend
892    where
893        TFilter: SetCursor + WithPartition,
894        TResponse: Send + 'static,
895    {
896        let mut streams = SelectAll::new();
897        for partition in 0..num_partitions {
898            let part_filter = filter.with_partition(Partition::new(partition + 1, num_partitions));
899            streams.push(self.filter_all_stream(part_filter).boxed_cond());
900        }
901
902        streams
903    }
904}
905
906/// Trait for resource types that allow filtering with fuzzy search.
907pub trait SearchItems<'a, TFilter, TSearch, TResponse>
908where
909    TFilter: Serialize + Sync + Send + 'a,
910    TSearch: Serialize + Sync + Send + 'a,
911    TResponse: Serialize + DeserializeOwned,
912    Self: WithApiClient + WithBasePath + Sync,
913{
914    /// Fuzzy search resources.
915    ///
916    /// # Arguments
917    ///
918    /// * `filter` - Simple filter applied to items.
919    /// * `search` - Fuzzy search.
920    /// * `limit` - Maximum number of items to retrieve.
921    fn search(
922        &'a self,
923        filter: TFilter,
924        search: TSearch,
925        limit: Option<u32>,
926    ) -> impl Future<Output = Result<Vec<TResponse>>> + CondSend {
927        async move {
928            let req = Search::<TFilter, TSearch>::new(filter, search, limit);
929            let response: ItemsVec<TResponse> = self
930                .get_client()
931                .post(&format!("{}/search", Self::BASE_PATH), &req)
932                .await?;
933            Ok(response.items)
934        }
935    }
936}
cognite/api/resource.rs

cognite/api/
resource.rs
