Move create document to pipeline architecture

sdk/core/src/headers/mod.rs

@@ -83,6 +83,13 @@ pub fn add_mandatory_header<T: AddAsHeader>(item: &T, builder: Builder) -> Build
     item.add_as_header(builder)
 }
 
+pub fn add_mandatory_header2<T: AddAsHeader>(
+    item: &T,
+    request: &mut crate::Request,
+) -> Result<(), crate::errors::HTTPHeaderError> {
+    item.add_as_header2(request)
+}
+
 pub const SERVER: &str = "server";
 pub const SOURCE_IF_MODIFIED_SINCE: &str = "x-ms-source-if-modified-since";
 pub const SOURCE_IF_UNMODIFIED_SINCE: &str = "x-ms-source-if-unmodified-since";

sdk/cosmos/examples/attachments_00.rs

@@ -1,3 +1,4 @@
+use azure_core::Context;
 use azure_cosmos::prelude::*;
 use serde::{Deserialize, Serialize};
 use std::borrow::Cow;
@@ -56,7 +57,10 @@ async fn main() -> Result<(), Box<dyn Error + Send + Sync>> {
     };
 
     // let's add an entity.
-    match client.create_document().execute(&doc).await {
+    match client
+        .create_document(Context::new(), &doc, CreateDocumentOptions::new())
+        .await
+    {
         Ok(_) => {
             println!("document created");
         }

sdk/cosmos/examples/database_00.rs

@@ -1,3 +1,4 @@
+use azure_core::Context;
 use azure_cosmos::prelude::*;
 use serde_json::Value;
 use std::error::Error;
@@ -40,10 +41,12 @@ async fn main() -> Result<(), Box<dyn Error + Send + Sync>> {
                 }"#;
                 let document: Value = serde_json::from_str(data)?;
 
-                let resp = collection_client
-                    .create_document()
+                let options = CreateDocumentOptions::new()
                     .is_upsert(true)
-                    .execute_with_partition_key(&document, &43u32)
+                    .partition_key(&43u32)
+                    .unwrap();
+                let resp = collection_client
+                    .create_document(Context::new(), &document, options)
                     .await?;
 
                 println!("resp == {:?}", resp);

sdk/cosmos/examples/document_00.rs

@@ -130,7 +130,9 @@ async fn main() -> Result<(), Box<dyn Error + Send + Sync>> {
     // The method create_document will return, upon success,
     // the document attributes.
 
-    let create_document_response = collection_client.create_document().execute(&doc).await?;
+    let create_document_response = collection_client
+        .create_document(Context::new(), &doc, CreateDocumentOptions::new())
+        .await?;
     println!(
         "create_document_response == {:#?}",
         create_document_response

sdk/cosmos/examples/document_entries_00.rs

@@ -59,7 +59,11 @@ async fn main() -> Result<(), Box<dyn Error + Send + Sync>> {
         };
 
         // let's add an entity.
-        response = Some(client.create_document().execute(&doc).await?);
+        response = Some(
+            client
+                .create_document(Context::new(), &doc, CreateDocumentOptions::new())
+                .await?,
+        );
     }
 
     println!("Created 5 documents.");

sdk/cosmos/examples/document_entries_01.rs

@@ -1,3 +1,4 @@
+use azure_core::Context;
 use azure_cosmos::prelude::*;
 use serde::{Deserialize, Serialize};
 use std::borrow::Cow;
@@ -49,9 +50,11 @@ async fn main() -> Result<(), Box<dyn Error + Send + Sync>> {
 
     // let's add an entity.
     let create_document_response = client
-        .create_document()
-        .is_upsert(true)
-        .execute(&doc)
+        .create_document(
+            Context::new(),
+            &doc,
+            CreateDocumentOptions::new().is_upsert(true),
+        )
         .await?;
 
     println!(

sdk/cosmos/examples/readme.rs

@@ -1,3 +1,4 @@
+use azure_core::Context;
 use serde::{Deserialize, Serialize};
 // Using the prelude module of the Cosmos crate makes easier to use the Rust Azure SDK for Cosmos.
 use azure_cosmos::prelude::*;
@@ -79,9 +80,11 @@ async fn main() -> Result<(), Box<dyn Error + Send + Sync>> {
         // insert it and store the returned session token for later use!
         session_token = Some(
             collection_client
-                .create_document()
-                .is_upsert(true) // this option will overwrite a preexisting document (if any)
-                .execute(&document_to_insert)
+                .create_document(
+                    Context::new(),
+                    &document_to_insert,
+                    CreateDocumentOptions::new().is_upsert(true),
+                )
                 .await?
                 .session_token, // get only the session token, if everything else was ok!
         );

sdk/cosmos/examples/user_permission_token.rs

@@ -115,9 +115,14 @@ async fn main() -> Result<(), Box<dyn Error + Send + Sync>> {
         .clone()
         .into_database_client(database_name.clone())
         .into_collection_client(collection_name.clone())
-        .create_document()
-        .is_upsert(true)
-        .execute_with_partition_key(&document, &"Gianluigi Bombatomica")
+        .create_document(
+            Context::new(),
+            &document,
+            CreateDocumentOptions::new()
+                .is_upsert(true)
+                .partition_key(&"Gianluigi Bombatomica")
+                .unwrap(),
+        )
         .await
     {
         Ok(_) => panic!("this should not happen!"),
@@ -154,9 +159,14 @@ async fn main() -> Result<(), Box<dyn Error + Send + Sync>> {
     let create_document_response = client
         .into_database_client(database_name)
         .into_collection_client(collection_name)
-        .create_document()
-        .is_upsert(true)
-        .execute_with_partition_key(&document, &"Gianluigi Bombatomica")
+        .create_document(
+            Context::new(),
+            &document,
+            CreateDocumentOptions::new()
+                .is_upsert(true)
+                .partition_key(&"Gianluigi Bombatomica")
+                .unwrap(),
+        )
         .await?;
     println!(
         "create_document_response == {:#?}",

sdk/cosmos/src/clients/collection_client.rs

@@ -1,9 +1,13 @@
 use super::{DatabaseClient, UserDefinedFunctionClient};
+use crate::authorization_policy::CosmosContext;
 use crate::clients::*;
+use crate::operations::*;
 use crate::requests;
 use crate::resources::ResourceType;
+use crate::CosmosEntity;
 use crate::ReadonlyString;
-use azure_core::HttpClient;
+use azure_core::PipelineContext;
+use azure_core::{pipeline::Pipeline, Context, HttpClient, Request};
 use serde::Serialize;
 
 /// A client for Cosmos collection resources.
@@ -60,8 +64,24 @@ impl CollectionClient {
     }
 
     /// create a document in a collection
-    pub fn create_document(&self) -> requests::CreateDocumentBuilder<'_, '_> {
-        requests::CreateDocumentBuilder::new(self)
+    pub async fn create_document<'a, D: Serialize + CosmosEntity<'a>>(
+        &self,
+        ctx: Context,
+        document: &'a D,
+        options: CreateDocumentOptions<'_>,
+    ) -> Result<CreateDocumentResponse, crate::Error> {
+        let mut request = self.prepare_request_pipeline(http::Method::POST);
+        let mut pipeline_context = PipelineContext::new(ctx, ResourceType::Documents.into());
+
+        options.decorate_request(&mut request, document)?;
+        let response = self
+            .pipeline()
+            .send(&mut pipeline_context, &mut request)
+            .await?
+            .validate(http::StatusCode::CREATED)
+            .await?;
+
+        Ok(CreateDocumentResponse::try_from(response).await?)
     }
 
     /// query documents in a collection
@@ -137,4 +157,18 @@ impl CollectionClient {
     pub(crate) fn http_client(&self) -> &dyn HttpClient {
         self.cosmos_client().http_client()
     }
+
+    pub(crate) fn pipeline(&self) -> &Pipeline<CosmosContext> {
+        self.cosmos_client().pipeline()
+    }
+
+    fn prepare_request_pipeline(&self, http_method: http::Method) -> Request {
+        let path = &format!(
+            "dbs/{}/colls/{}",
+            self.database_client().database_name(),
+            self.collection_name()
+        );
+        self.cosmos_client()
+            .prepare_request_pipeline(&path, http_method)
+    }
 }

sdk/cosmos/src/consistency_level.rs

@@ -1,5 +1,5 @@
 use crate::headers;
-use crate::operations::CreateUserResponse;
+use crate::operations::*;
 use crate::responses::*;
 use azure_core::AddAsHeader;
 use http::request;

sdk/cosmos/src/cosmos_entity.rs

@@ -1,4 +1,5 @@
 use crate::headers;
+use azure_core::Request as HttpRequest;
 use http::request::Builder;
 use serde::Serialize;
 
@@ -11,27 +12,20 @@ pub trait CosmosEntity<'a> {
     fn partition_key(&'a self) -> Self::Entity;
 }
 
+impl<'a> CosmosEntity<'a> for serde_json::Value {
+    type Entity = &'a Self;
+    fn partition_key(&'a self) -> Self::Entity {
+        self
+    }
+}
+
 /// Serialize the partition key in the format CosmosDB expects.
 pub(crate) fn serialize_partition_key<PK: Serialize>(pk: &PK) -> Result<String, serde_json::Error> {
     // this must be serialized as an array even tough CosmosDB supports only a sigle
     // partition key.
     serde_json::to_string(&[pk])
 }
 
-// Here we do not implement add_as_header because the trait does not support errors and serializing
-// with serde_json returns a Result. I am not sure why a serialization could fail (memory
-// allocation)? In case we are confident that no errors should arise we can implement the trait and just
-// unwrap the result of serde_json::to_string.
-pub(crate) fn add_as_partition_key_header<'a, P: CosmosEntity<'a>>(
-    pk: &'a P,
-    builder: Builder,
-) -> Result<Builder, serde_json::Error> {
-    Ok(builder.header(
-        headers::HEADER_DOCUMENTDB_PARTITIONKEY,
-        &serialize_partition_key(&pk.partition_key())?,
-    ))
-}
-
 pub(crate) fn add_as_partition_key_header_serialized(
     partition_key_serialized: &str,
     builder: Builder,
@@ -41,3 +35,13 @@ pub(crate) fn add_as_partition_key_header_serialized(
         partition_key_serialized,
     )
 }
+
+pub(crate) fn add_as_partition_key_header_serialized2(
+    partition_key_serialized: &str,
+    request: &mut HttpRequest,
+) {
+    request.headers_mut().insert(
+        headers::HEADER_DOCUMENTDB_PARTITIONKEY,
+        http::header::HeaderValue::from_str(partition_key_serialized).unwrap(),
+    );
+}

sdk/cosmos/src/lib.rs

@@ -10,6 +10,7 @@ should also be possible with this crate.
 ```no_run
 // Using the prelude module of the Cosmos crate makes easier to use the Rust Azure SDK for Cosmos DB.
 use azure_cosmos::prelude::*;
+use azure_core::Context;
 use serde::{Deserialize, Serialize};
 use std::error::Error;
 
@@ -75,9 +76,11 @@ async fn main() -> Result<(), Box<dyn Error + Send + Sync>> {
 
         // insert it
         collection_client
-            .create_document()
-            .is_upsert(true) // this option will overwrite a preexisting document (if any)
-            .execute(&document_to_insert)
+            .create_document(
+                Context::new(),
+                &document_to_insert,
+                CreateDocumentOptions::new().is_upsert(true),
+            )
             .await?;
     }
     // wow that was easy and fast, wasn't it? :)