Move create, get, and replace user to pipeline architecture (Azure#332)

* Move create, get, and replace User to pipeline architecture

* Updating e2e tests

* fix import

Co-authored-by: Joshua Gendein <[email protected]>

sdk/cosmos/examples/permission_00.rs

@@ -54,7 +54,9 @@ async fn main() -> Result<(), Box<dyn Error + Send + Sync>> {
         get_collection2_response
     );
 
-    let create_user_response = user_client.create_user().execute().await?;
+    let create_user_response = user_client
+        .create_user(Context::new(), CreateUserOptions::default())
+        .await?;
     println!("create_user_response == {:#?}", create_user_response);
 
     // create the first permission!

sdk/cosmos/examples/user_00.rs

@@ -1,3 +1,4 @@
+use azure_core::Context;
 use azure_cosmos::prelude::*;
 use std::error::Error;
 
@@ -27,18 +28,24 @@ async fn main() -> Result<(), Box<dyn Error + Send + Sync>> {
     let database_client = client.into_database_client(database_name);
     let user_client = database_client.clone().into_user_client(user_name.clone());
 
-    let create_user_response = user_client.create_user().execute().await?;
+    let create_user_response = user_client
+        .create_user(Context::new(), CreateUserOptions::new())
+        .await?;
     println!("create_user_response == {:#?}", create_user_response);
 
     let list_users_response = database_client.list_users().execute().await?;
     println!("list_users_response == {:#?}", list_users_response);
 
-    let get_user_response = user_client.get_user().execute().await?;
+    let get_user_response = user_client
+        .get_user(Context::new(), GetUserOptions::new())
+        .await?;
     println!("get_user_response == {:#?}", get_user_response);
 
     let new_user = format!("{}replaced", user_name);
 
-    let replace_user_response = user_client.replace_user().execute(&new_user).await?;
+    let replace_user_response = user_client
+        .replace_user(Context::new(), &new_user, ReplaceUserOptions::new())
+        .await?;
     println!("replace_user_response == {:#?}", replace_user_response);
 
     let user_client = database_client.into_user_client(new_user);

sdk/cosmos/examples/user_permission_token.rs

@@ -1,3 +1,4 @@
+use azure_core::Context;
 use azure_cosmos::prelude::*;
 use std::error::Error;
 
@@ -36,7 +37,9 @@ async fn main() -> Result<(), Box<dyn Error + Send + Sync>> {
     let get_collection_response = collection_client.get_collection().execute().await?;
     println!("get_collection_response == {:#?}", get_collection_response);
 
-    let create_user_response = user_client.create_user().execute().await?;
+    let create_user_response = user_client
+        .create_user(Context::new(), CreateUserOptions::default())
+        .await?;
     println!("create_user_response == {:#?}", create_user_response);
 
     // test list documents

sdk/cosmos/src/clients/user_client.rs

@@ -1,7 +1,10 @@
 use super::*;
+use crate::authorization_policy::CosmosContext;
+use crate::prelude::*;
 use crate::resources::ResourceType;
 use crate::{requests, ReadonlyString};
-use azure_core::HttpClient;
+use azure_core::PipelineContext;
+use azure_core::{pipeline::Pipeline, Context, HttpClient};
 
 /// A client for Cosmos user resources.
 #[derive(Debug, Clone)]
@@ -37,18 +40,81 @@ impl UserClient {
     }
 
     /// Create the user
-    pub fn create_user(&self) -> requests::CreateUserBuilder<'_, '_> {
-        requests::CreateUserBuilder::new(self)
+    pub async fn create_user(
+        &self,
+        ctx: Context,
+        options: CreateUserOptions,
+    ) -> Result<CreateUserResponse, crate::Error> {
+        let mut request = self.cosmos_client().prepare_request_pipeline(
+            &format!("dbs/{}/users", self.database_client.database_name()),
+            http::Method::POST,
+        );
+        let mut pipeline_context = PipelineContext::new(ctx, ResourceType::Users.into());
+
+        options.decorate_request(&mut request, self.user_name())?;
+        let response = self
+            .pipeline()
+            .send(&mut pipeline_context, &mut request)
+            .await?
+            .validate(http::StatusCode::CREATED)
+            .await?;
+
+        Ok(CreateUserResponse::try_from(response).await?)
     }
 
     /// Get the user
-    pub fn get_user(&self) -> requests::GetUserBuilder<'_, '_> {
-        requests::GetUserBuilder::new(self)
+    pub async fn get_user(
+        &self,
+        ctx: Context,
+        options: GetUserOptions,
+    ) -> Result<GetUserResponse, crate::Error> {
+        let mut request = self.cosmos_client().prepare_request_pipeline(
+            &format!(
+                "dbs/{}/users/{}",
+                self.database_client.database_name(),
+                self.user_name
+            ),
+            http::Method::GET,
+        );
+        let mut pipeline_context = PipelineContext::new(ctx, ResourceType::Users.into());
+
+        options.decorate_request(&mut request)?;
+        let response = self
+            .pipeline()
+            .send(&mut pipeline_context, &mut request)
+            .await?
+            .validate(http::StatusCode::OK)
+            .await?;
+
+        Ok(GetUserResponse::try_from(response).await?)
     }
 
     /// Replace the user
-    pub fn replace_user(&self) -> requests::ReplaceUserBuilder<'_, '_> {
-        requests::ReplaceUserBuilder::new(self)
+    pub async fn replace_user<S: AsRef<str>>(
+        &self,
+        ctx: Context,
+        user_name: S,
+        options: ReplaceUserOptions,
+    ) -> Result<ReplaceUserResponse, crate::Error> {
+        let mut request = self.cosmos_client().prepare_request_pipeline(
+            &format!(
+                "dbs/{}/users/{}",
+                self.database_client.database_name(),
+                self.user_name
+            ),
+            http::Method::PUT,
+        );
+        let mut pipeline_context = PipelineContext::new(ctx, ResourceType::Users.into());
+
+        options.decorate_request(&mut request, user_name.as_ref())?;
+        let response = self
+            .pipeline()
+            .send(&mut pipeline_context, &mut request)
+            .await?
+            .validate(http::StatusCode::OK)
+            .await?;
+
+        Ok(ReplaceUserResponse::try_from(response).await?)
     }
 
     /// Delete the user
@@ -69,14 +135,6 @@ impl UserClient {
         PermissionClient::new(self, permission_name)
     }
 
-    pub(crate) fn prepare_request(&self, method: http::Method) -> http::request::Builder {
-        self.cosmos_client().prepare_request(
-            &format!("dbs/{}/users", self.database_client().database_name()),
-            method,
-            ResourceType::Users,
-        )
-    }
-
     pub(crate) fn prepare_request_with_user_name(
         &self,
         method: http::Method,
@@ -95,4 +153,8 @@ impl UserClient {
     pub(crate) fn http_client(&self) -> &dyn HttpClient {
         self.cosmos_client().http_client()
     }
+
+    fn pipeline(&self) -> &Pipeline<CosmosContext> {
+        self.cosmos_client().pipeline()
+    }
 }

sdk/cosmos/src/consistency_level.rs

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

sdk/cosmos/src/operations/create_user.rs

@@ -0,0 +1,64 @@
+use crate::headers::from_headers::*;
+use crate::prelude::*;
+use azure_core::{
+    collect_pinned_stream,
+    headers::{etag_from_headers, session_token_from_headers},
+    Request as HttpRequest, Response as HttpResponse,
+};
+
+#[derive(Debug, Clone, Default)]
+pub struct CreateUserOptions {
+    consistency_level: Option<ConsistencyLevel>,
+}
+
+impl CreateUserOptions {
+    pub fn new() -> Self {
+        Self {
+            consistency_level: None,
+        }
+    }
+
+    setters! {
+        consistency_level: ConsistencyLevel => Some(consistency_level),
+    }
+
+    pub(crate) fn decorate_request(
+        &self,
+        request: &mut HttpRequest,
+        user_name: &str,
+    ) -> Result<(), crate::Error> {
+        azure_core::headers::add_optional_header2(&self.consistency_level, request)?;
+        let body = CreateUserBody { id: user_name };
+        request.set_body(bytes::Bytes::from(serde_json::to_string(&body)?).into());
+        Ok(())
+    }
+}
+
+#[derive(Serialize, Debug)]
+struct CreateUserBody<'a> {
+    id: &'a str,
+}
+
+#[derive(Debug, Clone)]
+pub struct CreateUserResponse {
+    pub user: User,
+    pub charge: f64,
+    pub activity_id: uuid::Uuid,
+    pub etag: String,
+    pub session_token: String,
+}
+
+impl CreateUserResponse {
+    pub async fn try_from(response: HttpResponse) -> Result<Self, crate::Error> {
+        let (_status_code, headers, pinned_stream) = response.deconstruct();
+        let body = collect_pinned_stream(pinned_stream).await?;
+
+        Ok(Self {
+            user: serde_json::from_slice(&body)?,
+            charge: request_charge_from_headers(&headers)?,
+            activity_id: activity_id_from_headers(&headers)?,
+            session_token: session_token_from_headers(&headers)?,
+            etag: etag_from_headers(&headers)?,
+        })
+    }
+}

sdk/cosmos/src/operations/get_user.rs

@@ -0,0 +1,55 @@
+use crate::headers::from_headers::*;
+use crate::prelude::*;
+use azure_core::{
+    collect_pinned_stream,
+    headers::{etag_from_headers, session_token_from_headers},
+    Request as HttpRequest, Response as HttpResponse,
+};
+
+#[derive(Debug, Clone, Default)]
+pub struct GetUserOptions {
+    consistency_level: Option<ConsistencyLevel>,
+}
+
+impl GetUserOptions {
+    pub fn new() -> Self {
+        Self {
+            consistency_level: None,
+        }
+    }
+
+    setters! {
+        consistency_level: ConsistencyLevel => Some(consistency_level),
+    }
+
+    pub(crate) fn decorate_request(&self, request: &mut HttpRequest) -> Result<(), crate::Error> {
+        azure_core::headers::add_optional_header2(&self.consistency_level, request)?;
+        request.set_body(bytes::Bytes::from_static(&[]).into());
+
+        Ok(())
+    }
+}
+
+#[derive(Debug, Clone)]
+pub struct GetUserResponse {
+    pub user: User,
+    pub charge: f64,
+    pub activity_id: uuid::Uuid,
+    pub etag: String,
+    pub session_token: String,
+}
+
+impl GetUserResponse {
+    pub async fn try_from(response: HttpResponse) -> Result<Self, crate::Error> {
+        let (_status_code, headers, pinned_stream) = response.deconstruct();
+        let body = collect_pinned_stream(pinned_stream).await?;
+
+        Ok(Self {
+            user: serde_json::from_slice(&body)?,
+            charge: request_charge_from_headers(&headers)?,
+            activity_id: activity_id_from_headers(&headers)?,
+            session_token: session_token_from_headers(&headers)?,
+            etag: etag_from_headers(&headers)?,
+        })
+    }
+}

sdk/cosmos/src/operations/mod.rs

@@ -4,8 +4,14 @@
 
 mod create_collection;
 mod create_database;
+mod create_user;
 mod get_database;
+mod get_user;
+mod replace_user;
 
 pub use create_collection::*;
 pub use create_database::*;
+pub use create_user::*;
 pub use get_database::*;
+pub use get_user::*;
+pub use replace_user::*;

sdk/cosmos/src/operations/replace_user.rs

@@ -0,0 +1,65 @@
+use crate::headers::from_headers::*;
+use crate::prelude::*;
+use azure_core::{
+    collect_pinned_stream,
+    headers::{etag_from_headers, session_token_from_headers},
+    Request as HttpRequest, Response as HttpResponse,
+};
+
+#[derive(Debug, Clone, Default)]
+pub struct ReplaceUserOptions {
+    consistency_level: Option<ConsistencyLevel>,
+}
+
+impl ReplaceUserOptions {
+    pub fn new() -> Self {
+        Self {
+            consistency_level: None,
+        }
+    }
+
+    setters! {
+        consistency_level: ConsistencyLevel => Some(consistency_level),
+    }
+
+    pub(crate) fn decorate_request(
+        &self,
+        request: &mut HttpRequest,
+        user_name: &str,
+    ) -> Result<(), crate::Error> {
+        azure_core::headers::add_optional_header2(&self.consistency_level, request)?;
+        let body = ReplaceUserBody { id: user_name };
+        request.set_body(bytes::Bytes::from(serde_json::to_string(&body)?).into());
+
+        Ok(())
+    }
+}
+
+#[derive(Debug, Clone)]
+pub struct ReplaceUserResponse {
+    pub user: User,
+    pub charge: f64,
+    pub activity_id: uuid::Uuid,
+    pub etag: String,
+    pub session_token: String,
+}
+
+impl ReplaceUserResponse {
+    pub async fn try_from(response: HttpResponse) -> Result<Self, crate::Error> {
+        let (_status_code, headers, pinned_stream) = response.deconstruct();
+        let body = collect_pinned_stream(pinned_stream).await?;
+
+        Ok(Self {
+            user: serde_json::from_slice(&body)?,
+            charge: request_charge_from_headers(&headers)?,
+            activity_id: activity_id_from_headers(&headers)?,
+            session_token: session_token_from_headers(&headers)?,
+            etag: etag_from_headers(&headers)?,
+        })
+    }
+}
+
+#[derive(Serialize)]
+struct ReplaceUserBody<'a> {
+    id: &'a str,
+}