Merge pull request #35 from meilisearch/retrieve-documents

implemement retrieve documents

src/data/search.rs

@@ -2,7 +2,7 @@ use std::collections::HashSet;
 use std::mem;
 use std::time::Instant;
 
-use anyhow::bail;
+use anyhow::{bail, Context};
 use meilisearch_tokenizer::{Analyzer, AnalyzerConfig};
 use milli::{Index, obkv_to_json, FacetCondition};
 use serde::{Deserialize, Serialize};
@@ -70,7 +70,7 @@ impl SearchQuery {
         let highlighter = Highlighter::new(&stop_words);
 
         for (_id, obkv) in index.documents(&rtxn, documents_ids).unwrap() {
-            let mut object = obkv_to_json(&displayed_fields, &fields_ids_map, obkv).unwrap();
+            let mut object = obkv_to_json(&displayed_fields, &fields_ids_map, obkv)?;
             if let Some(ref attributes_to_highlight) = self.attributes_to_highlight {
                 highlighter.highlight_record(&mut object, &found_words, attributes_to_highlight);
             }
@@ -165,4 +165,93 @@ impl Data {
             None => bail!("index {:?} doesn't exists", index.as_ref()),
         }
     }
+
+    pub async fn retrieve_documents<S>(
+        &self,
+        index: impl AsRef<str> + Send + Sync + 'static,
+        offset: usize,
+        limit: usize,
+        attributes_to_retrieve: Option<Vec<S>>,
+    ) -> anyhow::Result<Vec<Map<String, Value>>>
+    where
+        S: AsRef<str> + Send + Sync + 'static
+    {
+        let index_controller = self.index_controller.clone();
+        let documents: anyhow::Result<_> = tokio::task::spawn_blocking(move || {
+            let index = index_controller
+                .index(&index)?
+                .with_context(|| format!("Index {:?} doesn't exist", index.as_ref()))?;
+
+            let txn = index.read_txn()?;
+
+            let fields_ids_map = index.fields_ids_map(&txn)?;
+
+            let attributes_to_retrieve_ids = match attributes_to_retrieve {
+                Some(attrs) => attrs
+                    .iter()
+                    .filter_map(|f| fields_ids_map.id(f.as_ref()))
+                    .collect::<Vec<_>>(),
+                None => fields_ids_map.iter().map(|(id, _)| id).collect(),
+            };
+
+            let iter = index.documents.range(&txn, &(..))?
+                .skip(offset)
+                .take(limit);
+
+            let mut documents = Vec::new();
+
+            for entry in iter {
+                let (_id, obkv) = entry?;
+                let object = obkv_to_json(&attributes_to_retrieve_ids, &fields_ids_map, obkv)?;
+                documents.push(object);
+            }
+
+            Ok(documents)
+        }).await?;
+        documents
+    }
+
+    pub async fn retrieve_document<S>(
+        &self,
+        index: impl AsRef<str> + Sync + Send + 'static,
+        document_id: impl AsRef<str> + Sync + Send + 'static,
+        attributes_to_retrieve: Option<Vec<S>>,
+    ) -> anyhow::Result<Map<String, Value>>
+    where
+        S: AsRef<str> + Sync + Send + 'static,
+    {
+        let index_controller = self.index_controller.clone();
+        let document: anyhow::Result<_> = tokio::task::spawn_blocking(move || {
+            let index = index_controller
+                .index(&index)?
+                .with_context(|| format!("Index {:?} doesn't exist", index.as_ref()))?;
+            let txn = index.read_txn()?;
+
+            let fields_ids_map = index.fields_ids_map(&txn)?;
+
+            let attributes_to_retrieve_ids = match attributes_to_retrieve {
+                Some(attrs) => attrs
+                    .iter()
+                    .filter_map(|f| fields_ids_map.id(f.as_ref()))
+                    .collect::<Vec<_>>(),
+                None => fields_ids_map.iter().map(|(id, _)| id).collect(),
+            };
+
+            let internal_id = index
+                .external_documents_ids(&txn)?
+                .get(document_id.as_ref().as_bytes())
+                .with_context(|| format!("Document with id {} not found", document_id.as_ref()))?;
+
+            let document = index.documents(&txn, std::iter::once(internal_id))?
+                .into_iter()
+                .next()
+                .map(|(_, d)| d);
+
+            match document {
+                Some(document) => Ok(obkv_to_json(&attributes_to_retrieve_ids, &fields_ids_map, document)?),
+                None => bail!("Document with id {} not found", document_id.as_ref()),
+            }
+        }).await?;
+        document
+    }
 }

src/routes/document.rs

@@ -12,6 +12,9 @@ use crate::error::ResponseError;
 use crate::helpers::Authentication;
 use crate::routes::IndexParam;
 
+const DEFAULT_RETRIEVE_DOCUMENTS_OFFSET: usize = 0;
+const DEFAULT_RETRIEVE_DOCUMENTS_LIMIT: usize = 20;
+
 macro_rules! guard_content_type {
     ($fn_name:ident, $guard_value:literal) => {
         fn $fn_name(head: &actix_web::dev::RequestHead) -> bool {
@@ -49,10 +52,21 @@ pub fn services(cfg: &mut web::ServiceConfig) {
     wrap = "Authentication::Public"
 )]
 async fn get_document(
-    _data: web::Data<Data>,
-    _path: web::Path<DocumentParam>,
+    data: web::Data<Data>,
+    path: web::Path<DocumentParam>,
 ) -> Result<HttpResponse, ResponseError> {
-    todo!()
+    let index = path.index_uid.clone();
+    let id = path.document_id.clone();
+    match data.retrieve_document(index, id, None as Option<Vec<String>>).await {
+        Ok(document) => {
+            let json = serde_json::to_string(&document).unwrap();
+            Ok(HttpResponse::Ok().body(json))
+        }
+        Err(e) => {
+            error!("{}", e);
+            unimplemented!()
+        }
+    }
 }
 
 #[delete(
@@ -78,18 +92,36 @@ async fn delete_document(
 #[derive(Deserialize)]
 #[serde(rename_all = "camelCase", deny_unknown_fields)]
 struct BrowseQuery {
-    _offset: Option<usize>,
-    _limit: Option<usize>,
-    _attributes_to_retrieve: Option<String>,
+    offset: Option<usize>,
+    limit: Option<usize>,
+    attributes_to_retrieve: Option<String>,
 }
 
 #[get("/indexes/{index_uid}/documents", wrap = "Authentication::Public")]
 async fn get_all_documents(
-    _data: web::Data<Data>,
-    _path: web::Path<IndexParam>,
-    _params: web::Query<BrowseQuery>,
+    data: web::Data<Data>,
+    path: web::Path<IndexParam>,
+    params: web::Query<BrowseQuery>,
 ) -> Result<HttpResponse, ResponseError> {
-    todo!()
+    let attributes_to_retrieve = params
+        .attributes_to_retrieve
+        .as_ref()
+        .map(|attrs| attrs
+            .split(",")
+            .map(String::from)
+            .collect::<Vec<_>>());
+
+    match data.retrieve_documents(
+        path.index_uid.clone(),
+        params.offset.unwrap_or(DEFAULT_RETRIEVE_DOCUMENTS_OFFSET),
+        params.limit.unwrap_or(DEFAULT_RETRIEVE_DOCUMENTS_LIMIT),
+        attributes_to_retrieve).await {
+        Ok(docs) => {
+            let json = serde_json::to_string(&docs).unwrap();
+            Ok(HttpResponse::Ok().body(json))
+        }
+        Err(_) => { todo!() }
+    }
 }
 
 #[derive(Deserialize)]

src/routes/index.rs

@@ -32,7 +32,6 @@ async fn list_indexes(data: web::Data<Data>) -> Result<HttpResponse, ResponseErr
             unimplemented!()
         }
     }
-
 }
 
 #[get("/indexes/{index_uid}", wrap = "Authentication::Private")]