Added if_match_etag to Item Options

sdk/cosmos/azure_data_cosmos/CHANGELOG.md

@@ -17,6 +17,7 @@
 * Added a function `CosmosClient::with_connection_string` to enable `CosmosClient` creation via connection string. ([#2641](https://github.com/Azure/azure-sdk-for-rust/pull/2641))
 * Added support for executing limited cross-partition queries through the Gateway. See <https://learn.microsoft.com/rest/api/cosmos-db/querying-cosmosdb-resources-using-the-rest-api#queries-that-cannot-be-served-by-gateway> for more details on these limitations. ([#2577](https://github.com/Azure/azure-sdk-for-rust/pull/2577))
 * Added a preview feature (behind `preview_query_engine` feature flag) to allow the Rust SDK to integrate with an external query engine for performing cross-partition queries. ([#2577](https://github.com/Azure/azure-sdk-for-rust/pull/2577))
+* Added 'if_match_etag' to ItemOptions and necessary functions ([#2705](https://github.com/Azure/azure-sdk-for-rust/pull/2705))
 
 ### Breaking Changes
 

sdk/cosmos/azure_data_cosmos/src/clients/container_client.rs

@@ -12,7 +12,7 @@ use crate::{
 };
 
 use azure_core::http::{
-    headers,
+    headers::{self},
     request::{options::ContentType, Request},
     response::Response,
     Method,
@@ -358,6 +358,9 @@ impl ContainerClient {
         if !options.enable_content_response_on_write {
             req.insert_header(headers::PREFER, constants::PREFER_MINIMAL);
         }
+        if let Some(etag) = options.if_match_etag {
+            req.insert_header(headers::IF_MATCH, etag);
+        }
         req.insert_headers(&partition_key.into())?;
         req.insert_headers(&ContentType::APPLICATION_JSON)?;
         req.set_json(&item)?;
@@ -447,6 +450,9 @@ impl ContainerClient {
         if !options.enable_content_response_on_write {
             req.insert_header(headers::PREFER, constants::PREFER_MINIMAL);
         }
+        if let Some(etag) = options.if_match_etag {
+            req.insert_header(headers::IF_MATCH, etag);
+        }
         req.insert_header(constants::IS_UPSERT, "true");
         req.insert_headers(&partition_key.into())?;
         req.insert_headers(&ContentType::APPLICATION_JSON)?;
@@ -537,6 +543,9 @@ impl ContainerClient {
         let link = self.items_link.item(item_id);
         let url = self.pipeline.url(&link);
         let mut req = Request::new(url, Method::Delete);
+        if let Some(etag) = options.if_match_etag {
+            req.insert_header(headers::IF_MATCH, etag);
+        }
         req.insert_headers(&partition_key.into())?;
         self.pipeline
             .send(options.method_options.context, &mut req, link)
@@ -613,6 +622,9 @@ impl ContainerClient {
         if !options.enable_content_response_on_write {
             req.insert_header(headers::PREFER, constants::PREFER_MINIMAL);
         }
+        if let Some(etag) = options.if_match_etag {
+            req.insert_header(headers::IF_MATCH, etag);
+        }
         req.insert_headers(&partition_key.into())?;
         req.insert_headers(&ContentType::APPLICATION_JSON)?;
         req.set_json(&patch)?;

sdk/cosmos/azure_data_cosmos/src/options/mod.rs

@@ -1,7 +1,7 @@
 // Copyright (c) Microsoft Corporation. All rights reserved.
 // Licensed under the MIT License.
 
-use azure_core::http::{ClientMethodOptions, ClientOptions};
+use azure_core::http::{ClientMethodOptions, ClientOptions, Etag};
 
 use crate::models::ThroughputProperties;
 
@@ -47,7 +47,10 @@ pub struct DeleteDatabaseOptions<'a> {
 #[derive(Clone, Default)]
 pub struct ItemOptions<'a> {
     pub method_options: ClientMethodOptions<'a>,
-
+    /// IfMatchEtag is used to ensure optimistic concurrency control, it helps prevent accidental overwrites and maintains data integrity.
+    ///
+    /// https://learn.microsoft.com/en-us/azure/cosmos-db/nosql/database-transactions-optimistic-concurrency#optimistic-concurrency-control
+    pub if_match_etag: Option<Etag>,
     /// When this value is true, write operations will respond with the new value of the resource being written.
     ///
     /// The default for this is `false`, which reduces the network and CPU burden that comes from serializing and deserializing the response.

sdk/cosmos/azure_data_cosmos/tests/cosmos_items.rs

@@ -2,6 +2,7 @@
 
 mod framework;
 
+use azure_core::http::Etag;
 use azure_core_test::{recorded, TestContext};
 use azure_data_cosmos::{
     clients::ContainerClient,
@@ -201,6 +202,7 @@ pub async fn item_read_system_properties(context: TestContext) -> Result<(), Box
         read_item.get("_rid").is_some(),
         "expected _rid to be present"
     );
+
     assert!(
         read_item.get("_etag").is_some(),
         "expected _etag to be present"
@@ -405,3 +407,293 @@ pub async fn item_null_partition_key(context: TestContext) -> Result<(), Box<dyn
     account.cleanup().await?;
     Ok(())
 }
+
+#[recorded::test]
+pub async fn item_replace_if_match_etag(context: TestContext) -> Result<(), Box<dyn Error>> {
+    let account = TestAccount::from_env(context, None).await?;
+    let cosmos_client = account.connect_with_key(None)?;
+    let container_client = create_container(&account, &cosmos_client).await?;
+
+    //Create an item
+    let mut item = TestItem {
+        id: "Item1".into(),
+        partition_key: Some("Partition1".into()),
+        value: 42,
+        nested: NestedItem {
+            nested_value: "Nested".into(),
+        },
+        bool_value: true,
+    };
+
+    let response = container_client
+        .create_item("Partition1", &item, None)
+        .await?;
+
+    //Store Etag from response
+    let etag: Etag = response
+        .headers()
+        .get_str(&azure_core::http::headers::ETAG)
+        .expect("expected the etag to be returned")
+        .into();
+
+    //Replace item with correct Etag
+    item.value = 24;
+    item.nested.nested_value = "Updated".into();
+
+    container_client
+        .replace_item(
+            "Partition1",
+            "Item1",
+            &item,
+            Some(ItemOptions {
+                if_match_etag: Some(etag),
+                ..Default::default()
+            }),
+        )
+        .await?;
+
+    //Replace item with incorrect Etag
+    item.value = 52;
+    item.nested.nested_value = "UpdatedAgain".into();
+
+    let response = container_client
+        .replace_item(
+            "Partition1",
+            "Item1",
+            &item,
+            Some(ItemOptions {
+                if_match_etag: Some("incorrectEtag".into()),
+                ..Default::default()
+            }),
+        )
+        .await;
+
+    assert_eq!(
+        Some(azure_core::http::StatusCode::PreconditionFailed),
+        response
+            .expect_err("expected the server to return an error")
+            .http_status()
+    );
+
+    account.cleanup().await?;
+    Ok(())
+}
+
+#[recorded::test]
+pub async fn item_upsert_if_match_etag(context: TestContext) -> Result<(), Box<dyn Error>> {
+    let account = TestAccount::from_env(context, None).await?;
+    let cosmos_client = account.connect_with_key(None)?;
+    let container_client = create_container(&account, &cosmos_client).await?;
+
+    //Create an item
+    let mut item = TestItem {
+        id: "Item1".into(),
+        partition_key: Some("Partition1".into()),
+        value: 42,
+        nested: NestedItem {
+            nested_value: "Nested".into(),
+        },
+        bool_value: true,
+    };
+
+    let response = container_client
+        .create_item("Partition1", &item, None)
+        .await?;
+
+    //Store Etag from response
+    let etag: Etag = response
+        .headers()
+        .get_str(&azure_core::http::headers::ETAG)
+        .expect("expected the etag to be returned")
+        .into();
+
+    //Upsert item with correct Etag
+    item.value = 24;
+    item.nested.nested_value = "Updated".into();
+
+    container_client
+        .upsert_item(
+            "Partition1",
+            &item,
+            Some(ItemOptions {
+                if_match_etag: Some(etag),
+                ..Default::default()
+            }),
+        )
+        .await?;
+
+    //Upsert item with incorrect Etag
+    item.value = 52;
+    item.nested.nested_value = "UpdatedAgain".into();
+
+    let response = container_client
+        .upsert_item(
+            "Partition1",
+            &item,
+            Some(ItemOptions {
+                if_match_etag: Some("incorrectEtag".into()),
+                ..Default::default()
+            }),
+        )
+        .await;
+
+    assert_eq!(
+        Some(azure_core::http::StatusCode::PreconditionFailed),
+        response
+            .expect_err("expected the server to return an error")
+            .http_status()
+    );
+
+    account.cleanup().await?;
+    Ok(())
+}
+
+#[recorded::test]
+pub async fn item_delete_if_match_etag(context: TestContext) -> Result<(), Box<dyn Error>> {
+    let account = TestAccount::from_env(context, None).await?;
+    let cosmos_client = account.connect_with_key(None)?;
+    let container_client = create_container(&account, &cosmos_client).await?;
+
+    //Create an item
+    let item = TestItem {
+        id: "Item1".into(),
+        partition_key: Some("Partition1".into()),
+        value: 42,
+        nested: NestedItem {
+            nested_value: "Nested".into(),
+        },
+        bool_value: true,
+    };
+
+    let response = container_client
+        .create_item("Partition1", &item, None)
+        .await?;
+
+    //Store Etag from response
+    let etag: Etag = response
+        .headers()
+        .get_str(&azure_core::http::headers::ETAG)
+        .expect("expected the etag to be returned")
+        .into();
+
+    //Delete item with correct Etag
+    container_client
+        .delete_item(
+            "Partition1",
+            "Item1",
+            Some(ItemOptions {
+                if_match_etag: Some(etag),
+                ..Default::default()
+            }),
+        )
+        .await?;
+
+    //Add item again for second delete test
+    container_client
+        .create_item("Partition1", &item, None)
+        .await?;
+
+    //Delete item with incorrect Etag
+    let response = container_client
+        .delete_item(
+            "Partition1",
+            "Item1",
+            Some(ItemOptions {
+                if_match_etag: Some("incorrectEtag".into()),
+                ..Default::default()
+            }),
+        )
+        .await;
+
+    assert_eq!(
+        Some(azure_core::http::StatusCode::PreconditionFailed),
+        response
+            .expect_err("expected the server to return an error")
+            .http_status()
+    );
+
+    account.cleanup().await?;
+    Ok(())
+}
+
+#[recorded::test]
+pub async fn item_patch_if_match_etag(context: TestContext) -> Result<(), Box<dyn Error>> {
+    let account = TestAccount::from_env(context, None).await?;
+    let cosmos_client = account.connect_with_key(None)?;
+    let container_client = create_container(&account, &cosmos_client).await?;
+
+    //Create an item
+    let item = TestItem {
+        id: "Item1".into(),
+        partition_key: Some("Partition1".into()),
+        value: 42,
+        nested: NestedItem {
+            nested_value: "Nested".into(),
+        },
+        bool_value: true,
+    };
+
+    let response = container_client
+        .create_item("Partition1", &item, None)
+        .await?;
+
+    //Store Etag from response
+    let etag: Etag = response
+        .headers()
+        .get_str(&azure_core::http::headers::ETAG)
+        .expect("expected the etag to be returned")
+        .into();
+
+    //Patch item with correct Etag
+    let patch = PatchDocument::default()
+        .with_replace("/nested/nested_value", "Patched")?
+        .with_increment("/value", 10)?;
+
+    container_client
+        .patch_item(
+            "Partition1",
+            "Item1",
+            patch,
+            Some(ItemOptions {
+                if_match_etag: Some(etag),
+                ..Default::default()
+            }),
+        )
+        .await?;
+
+    let patched_item: TestItem = container_client
+        .read_item("Partition1", "Item1", None)
+        .await?
+        .into_body()
+        .await?;
+
+    assert_eq!("Patched", patched_item.nested.nested_value);
+    assert_eq!(52, patched_item.value);
+
+    //Patch item with incorrect Etag
+    let patch = PatchDocument::default()
+        .with_replace("/nested/nested_value", "PatchedIncorrect")?
+        .with_increment("/value", 15)?;
+
+    let response = container_client
+        .patch_item(
+            "Partition1",
+            "Item1",
+            patch,
+            Some(ItemOptions {
+                if_match_etag: Some("incorrectEtag".into()),
+                ..Default::default()
+            }),
+        )
+        .await;
+
+    assert_eq!(
+        Some(azure_core::http::StatusCode::PreconditionFailed),
+        response
+            .expect_err("expected the server to return an error")
+            .http_status()
+    );
+
+    account.cleanup().await?;
+    Ok(())
+}