Rust client

The OpenSearch Rust client lets you connect your Rust application with the data in your OpenSearch cluster. For the client’s complete API documentation and additional examples, see the OpenSearch docs.rs documentation.

This getting started guide illustrates how to connect to OpenSearch, index documents, and run queries. For the client source code, see the opensearch-rs repo.

Setup

If you’re starting a new project, add the opensearch crate to Cargo.toml:

  1. [dependencies]
  2. opensearch = "1.0.0"

copy

Additionally, you may want to add the following serde dependencies that help serialize types to JSON and deserialize JSON responses:

  1. serde = "~1"
  2. serde_json = "~1"

copy

The Rust client uses the higher-level reqwest HTTP client library for HTTP requests, and reqwest uses the tokio platform to support asynchronous requests. If you are planning to use asynchronous functions, you need to add the tokio dependency to Cargo.toml:

  1. tokio = { version = "*", features = ["full"] }

copy

See the Sample program section for the complete Cargo.toml file.

To use the Rust client API, import the modules, structs, and enums you need:

  1. use opensearch::OpenSearch;

copy

Connecting to OpenSearch

To connect to the default OpenSearch host, create a default client object that connects to OpenSearch at the address http://localhost:9200:

  1. let client = OpenSearch::default();

copy

To connect to an OpenSearch host that is running at a different address, create a client with the specified address:

  1. let transport = Transport::single_node("http://localhost:9200")?;
  2. let client = OpenSearch::new(transport);

copy

Alternatively, you can customize the URL and use a connection pool by creating a TransportBuilder struct and passing it to OpenSearch::new to create a new instance of the client:

  1. let url = Url::parse("http://localhost:9200")?;
  2. let conn_pool = SingleNodeConnectionPool::new(url);
  3. let transport = TransportBuilder::new(conn_pool).disable_proxy().build()?;
  4. let client = OpenSearch::new(transport);

copy

Creating an index

To create an OpenSearch index, use the create function of the opensearch::indices::Indices struct. You can use the following code to construct a JSON object with custom mappings:

  1. let response = client
  2.     .indices()
  3.     .create(IndicesCreateParts::Index("movies"))
  4.     .body(json!({
  5.         "mappings" : {
  6.             "properties" : {
  7.                 "title" : { "type" : "text" }
  8.             }
  9.         }
  10.     }))
  11.     .send()
  12.     .await?;

copy

Indexing a document

You can index a document into OpenSearch using the client’s index function:

  1. let response = client
  2.     .index(IndexParts::IndexId("movies", "1"))
  3.     .body(json!({
  4.         "id": 1,
  5.         "title": "Moneyball",
  6.         "director": "Bennett Miller",
  7.         "year": "2011"
  8.     }))
  9.     .send()
  10.     .await?;

copy

Performing bulk operations

You can perform several operations at the same time by using the client’s bulk function. First, create the JSON body of a Bulk API call, and then pass it to the bulk function:

  1. let mut body: Vec<JsonBody<_>> = Vec::with_capacity(4);
  3. // add the first operation and document
  4. body.push(json!({"index": {"_id": "2"}}).into());
  5. body.push(json!({
  6.     "id": 2,
  7.     "title": "Interstellar",
  8.     "director": "Christopher Nolan",
  9.     "year": "2014"
  10. }).into());
  12. // add the second operation and document
  13. body.push(json!({"index": {"_id": "3"}}).into());
  14. body.push(json!({
  15.     "id": 3,
  16.     "title": "Star Trek Beyond",
  17.     "director": "Justin Lin",
  18.     "year": "2015"
  19. }).into());
  21. let response = client
  22.     .bulk(BulkParts::Index("movies"))
  23.     .body(body)
  24.     .send()
  25.     .await?;

copy

Searching for documents

The easiest way to search for documents is to construct a query string. The following code uses a multi_match query to search for “miller” in the title and director fields. It boosts the documents where “miller” appears in the title field:

  1. response = client
  2.     .search(SearchParts::Index(&["movies"]))
  3.     .from(0)
  4.     .size(10)
  5.     .body(json!({
  6.         "query": {
  7.             "multi_match": {
  8.                 "query": "miller",
  9.                 "fields": ["title^2", "director"]
  10.             }           
  11.         }
  12.     }))
  13.     .send()
  14.     .await?;

copy

You can then read the response body as JSON and iterate over the hits array to read all the _source documents:

  1. let response_body = response.json::<Value>().await?;
  2. for hit in response_body["hits"]["hits"].as_array().unwrap() {
  3.     // print the source document
  4.     println!("{}", serde_json::to_string_pretty(&hit["_source"]).unwrap());
  5. }

copy

Deleting a document

You can delete a document using the client’s delete function:

  1. let response = client
  2.     .delete(DeleteParts::IndexId("movies", "2"))
  3.     .send()
  4.     .await?;

copy

Deleting an index

You can delete an index using the delete function of the opensearch::indices::Indices struct:

  1. let response = client
  2.     .indices()
  3.     .delete(IndicesDeleteParts::Index(&["movies"]))
  4.     .send()
  5.     .await?;

copy

Sample program

The sample program uses the following Cargo.toml file with all dependencies described in the Setup section:

  1. [package]
  2. name = "os_rust_project"
  3. version = "0.1.0"
  4. edition = "2021"
  6. # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
  8. [dependencies]
  9. opensearch = "1.0.0"
  10. tokio = { version = "*", features = ["full"] }
  11. serde = "~1"
  12. serde_json = "~1"

copy

The following sample program creates a client, adds an index with non-default mappings, inserts a document, performs bulk operations, searches for the document, deletes the document, and then deletes the index:

  1. use opensearch::{DeleteParts, OpenSearch, IndexParts, http::request::JsonBody, BulkParts, SearchParts};
  2. use opensearch::{indices::{IndicesDeleteParts, IndicesCreateParts}};
  3. use serde_json::{json, Value};
  5. #[tokio::main]
  6. async fn main() -> Result<(), Box<dyn std::error::Error>> {
  7.     let client = OpenSearch::default();
  9.     // Create an index
  10.     let mut response = client
  11.     .indices()
  12.     .create(IndicesCreateParts::Index("movies"))
  13.     .body(json!({
  14.         "mappings" : {
  15.             "properties" : {
  16.                 "title" : { "type" : "text" }
  17.             }
  18.         }
  19.     }))
  20.     .send()
  21.     .await?;
  23.     let mut successful = response.status_code().is_success();
  25.     if successful {
  26.         println!("Successfully created an index");
  27.     }
  28.     else {
  29.         println!("Could not create an index");
  30.     }
  32.     // Index a single document
  33.     println!("Indexing a single document...");
  34.     response = client
  35.     .index(IndexParts::IndexId("movies", "1"))
  36.     .body(json!({
  37.         "id": 1,
  38.         "title": "Moneyball",
  39.         "director": "Bennett Miller",
  40.         "year": "2011"
  41.     }))
  42.     .send()
  43.     .await?;
  45.     successful = response.status_code().is_success();
  47.     if successful {
  48.         println!("Successfully indexed a document");
  49.     }
  50.     else {
  51.         println!("Could not index document");
  52.     }
  54.     // Index multiple documents using the bulk operation
  56.     println!("Indexing multiple documents...");
  58.     let mut body: Vec<JsonBody<_>> = Vec::with_capacity(4);
  60.     // add the first operation and document
  61.     body.push(json!({"index": {"_id": "2"}}).into());
  62.     body.push(json!({
  63.         "id": 2,
  64.         "title": "Interstellar",
  65.         "director": "Christopher Nolan",
  66.         "year": "2014"
  67.     }).into());
  69.     // add the second operation and document
  70.     body.push(json!({"index": {"_id": "3"}}).into());
  71.     body.push(json!({
  72.         "id": 3,
  73.         "title": "Star Trek Beyond",
  74.         "director": "Justin Lin",
  75.         "year": "2015"
  76.     }).into());
  78.     response = client
  79.         .bulk(BulkParts::Index("movies"))
  80.         .body(body)
  81.         .send()
  82.         .await?;
  84.     let mut response_body = response.json::<Value>().await?;
  85.     successful = response_body["errors"].as_bool().unwrap() == false;
  87.     if successful {
  88.         println!("Successfully performed bulk operations");
  89.     }
  90.     else {
  91.         println!("Could not perform bulk operations");
  92.     }
  94.     // Search for a document
  96.     println!("Searching for a document...");
  97.     response = client
  98.     .search(SearchParts::Index(&["movies"]))
  99.     .from(0)
  100.     .size(10)
  101.     .body(json!({
  102.         "query": {
  103.             "multi_match": {
  104.                 "query": "miller",
  105.                 "fields": ["title^2", "director"]
  106.             }           
  107.         }
  108.     }))
  109.     .send()
  110.     .await?;
  112.     response_body = response.json::<Value>().await?;
  113.     for hit in response_body["hits"]["hits"].as_array().unwrap() {
  114.         // print the source document
  115.         println!("{}", serde_json::to_string_pretty(&hit["_source"]).unwrap());
  116.     }
  118.     // Delete a document
  120.     response = client
  121.     .delete(DeleteParts::IndexId("movies", "2"))
  122.     .send()
  123.     .await?;
  125.     successful = response.status_code().is_success();
  127.     if successful {
  128.         println!("Successfully deleted a document");
  129.     }
  130.     else {
  131.         println!("Could not delete document");
  132.     }
  134.     // Delete the index
  136.     response = client
  137.     .indices()
  138.     .delete(IndicesDeleteParts::Index(&["movies"]))
  139.     .send()
  140.     .await?;
  142.     successful = response.status_code().is_success();
  144.     if successful {
  145.         println!("Successfully deleted the index");
  146.     }
  147.     else {
  148.         println!("Could not delete the index");
  149.     }
  151.     Ok(())
  152. }

copy