Create ListDatabases resource

This converts from the older req/res-style API into the newer pipeline-based API (ref: #290).

Author: yoshuawuyts

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/collection.rs

@@ -33,7 +33,9 @@ async fn main() -> Result<(), Box<dyn Error + Send + Sync>> {
     // The Cosmos' client exposes a lot of methods. This one lists the databases in the specified
     // account. Database do not implement Display but deref to &str so you can pass it to methods
     // both as struct or id.
-    let databases = client.list_databases().execute().await?;
+    let databases = client
+        .list_databases(Context::new(), ListDatabasesOptions::new())
+        .await?;
 
     println!(
         "Account {} has {} database(s)",

sdk/cosmos/examples/create_delete_database.rs

@@ -34,7 +34,9 @@ async fn main() -> Result<(), Box<dyn Error + Send + Sync>> {
     // account. Database do not implement Display but deref to &str so you can pass it to methods
     // both as struct or id.
 
-    let list_databases_response = client.list_databases().execute().await?;
+    let list_databases_response = client
+        .list_databases(Context::new(), ListDatabasesOptions::new())
+        .await?;
     println!("list_databases_response = {:#?}", list_databases_response);
 
     let db = client

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;
@@ -14,7 +15,9 @@ async fn main() -> Result<(), Box<dyn Error + Send + Sync>> {
 
     let client = CosmosClient::new(account, authorization_token, CosmosOptions::default());
 
-    let dbs = client.list_databases().execute().await?;
+    let dbs = client
+        .list_databases(Context::new(), ListDatabasesOptions::new())
+        .await?;
 
     for db in dbs.databases {
         println!("database == {:?}", db);

sdk/cosmos/examples/document_00.rs

@@ -58,8 +58,7 @@ async fn main() -> Result<(), Box<dyn Error + Send + Sync>> {
     // specific azure_cosmos::Error. In this example we will look for a specific database
     // so we chain a filter operation.
     let db = client
-        .list_databases()
-        .execute()
+        .list_databases(Context::new(), ListDatabasesOptions::new())
         .await?
         .databases
         .into_iter()

sdk/cosmos/src/clients/cosmos_client.rs

@@ -4,7 +4,7 @@ use crate::headers::*;
 use crate::operations::*;
 use crate::resources::permission::AuthorizationToken;
 use crate::resources::ResourceType;
-use crate::{requests, ReadonlyString, TimeNonce};
+use crate::{ReadonlyString, TimeNonce};
 use azure_core::pipeline::Pipeline;
 use azure_core::HttpClient;
 use azure_core::Request;
@@ -163,8 +163,23 @@ impl CosmosClient {
     }
 
     /// List all databases
-    pub fn list_databases(&self) -> requests::ListDatabasesBuilder<'_> {
-        requests::ListDatabasesBuilder::new(self)
+    pub async fn list_databases(
+        &self,
+        ctx: Context,
+        options: ListDatabasesOptions,
+    ) -> Result<ListDatabasesResponse, crate::Error> {
+        let mut request = self.prepare_request_pipeline("dbs", http::Method::GET);
+        let mut pipeline_context = PipelineContext::new(ctx, ResourceType::Databases.into());
+
+        options.decorate_request(&mut request).await?;
+        let response = self
+            .pipeline()
+            .send(&mut pipeline_context, &mut request)
+            .await?
+            .validate(http::StatusCode::OK)
+            .await?;
+
+        Ok(ListDatabasesResponse::try_from(response).await?)
     }
 
     /// Convert into a [`DatabaseClient`]

sdk/cosmos/src/operations/list_databases.rs

@@ -0,0 +1,127 @@
+use crate::headers::from_headers::*;
+use crate::prelude::*;
+use crate::resources::Database;
+use crate::ResourceQuota;
+
+use azure_core::headers::{continuation_token_from_headers_optional, session_token_from_headers};
+use azure_core::{collect_pinned_stream, prelude::*, Request, Response};
+use chrono::{DateTime, Utc};
+
+#[derive(Debug, Clone)]
+pub struct ListDatabasesOptions {
+    consistency_level: Option<ConsistencyLevel>,
+    max_item_count: MaxItemCount,
+}
+
+impl ListDatabasesOptions {
+    pub fn new() -> Self {
+        Self {
+            consistency_level: None,
+            max_item_count: MaxItemCount::new(-1),
+        }
+    }
+
+    setters! {
+        consistency_level: ConsistencyLevel => Some(consistency_level),
+        max_item_count: i32 => MaxItemCount::new(max_item_count),
+    }
+
+    pub async fn decorate_request(&self, request: &mut Request) -> Result<(), crate::Error> {
+        azure_core::headers::add_optional_header2(&self.consistency_level, request)?;
+        azure_core::headers::add_mandatory_header2(&self.max_item_count, request)?;
+        Ok(())
+    }
+
+    // pub fn stream(&self) -> impl Stream<Item = Result<ListDatabasesResponse, crate::Error>> + '_ {
+    //     #[derive(Debug, Clone, PartialEq)]
+    //     enum States {
+    //         Init,
+    //         Continuation(String),
+    //     }
+
+    //     unfold(
+    //         Some(States::Init),
+    //         move |continuation_token: Option<States>| {
+    //             async move {
+    //                 debug!("continuation_token == {:?}", &continuation_token);
+    //                 let response = match continuation_token {
+    //                     Some(States::Init) => self.decorate_request().await,
+    //                     Some(States::Continuation(continuation_token)) => {
+    //                         self.clone()
+    //                             .continuation(continuation_token.as_str())
+    //                             .decorate_request()
+    //                             .await
+    //                     }
+    //                     None => return None,
+    //                 };
+
+    //                 // the ? operator does not work in async move (yet?)
+    //                 // so we have to resort to this boilerplate
+    //                 let response = match response {
+    //                     Ok(response) => response,
+    //                     Err(err) => return Some((Err(err), None)),
+    //                 };
+
+    //                 let continuation_token = response
+    //                     .continuation_token
+    //                     .as_ref()
+    //                     .map(|ct| States::Continuation(ct.to_owned()));
+
+    //                 Some((Ok(response), continuation_token))
+    //             }
+    //         },
+    //     )
+    // }
+}
+
+#[derive(Clone, PartialEq, PartialOrd, Debug)]
+pub struct ListDatabasesResponse {
+    pub rid: String,
+    pub databases: Vec<Database>,
+    pub count: u32,
+    pub activity_id: uuid::Uuid,
+    pub charge: f64,
+    pub session_token: String,
+    pub last_state_change: DateTime<Utc>,
+    pub resource_quota: Vec<ResourceQuota>,
+    pub resource_usage: Vec<ResourceQuota>,
+    pub schema_version: String,
+    pub service_version: String,
+    pub continuation_token: Option<String>,
+    pub gateway_version: String,
+}
+
+impl ListDatabasesResponse {
+    pub(crate) async fn try_from(response: Response) -> Result<Self, crate::Error> {
+        let (_status_code, headers, pinned_stream) = response.deconstruct();
+        let body = collect_pinned_stream(pinned_stream).await?;
+
+        #[derive(Deserialize, Debug)]
+        pub struct Response {
+            #[serde(rename = "_rid")]
+            rid: String,
+            #[serde(rename = "Databases")]
+            pub databases: Vec<Database>,
+            #[serde(rename = "_count")]
+            pub count: u32,
+        }
+
+        let response: Response = serde_json::from_slice(&body)?;
+
+        Ok(Self {
+            rid: response.rid,
+            databases: response.databases,
+            count: response.count,
+            charge: request_charge_from_headers(&headers)?,
+            activity_id: activity_id_from_headers(&headers)?,
+            session_token: session_token_from_headers(&headers)?,
+            last_state_change: last_state_change_from_headers(&headers)?,
+            resource_quota: resource_quota_from_headers(&headers)?,
+            resource_usage: resource_usage_from_headers(&headers)?,
+            schema_version: schema_version_from_headers(&headers)?.to_owned(),
+            service_version: service_version_from_headers(&headers)?.to_owned(),
+            continuation_token: continuation_token_from_headers_optional(&headers)?,
+            gateway_version: gateway_version_from_headers(&headers)?.to_owned(),
+        })
+    }
+}

sdk/cosmos/src/operations/mod.rs

@@ -7,11 +7,13 @@ mod create_database;
 mod create_user;
 mod get_database;
 mod get_user;
+mod list_databases;
 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 list_databases::*;
 pub use replace_user::*;

sdk/cosmos/src/requests/list_databases_builder.rs

@@ -30,7 +30,6 @@ mod get_partition_key_ranges_builder;
 mod get_permission_builer;
 mod list_attachments_builder;
 mod list_collections_builder;
-mod list_databases_builder;
 mod list_documents_builder;
 mod list_permissions_builder;
 mod list_stored_procedures_builder;
@@ -69,7 +68,6 @@ pub use get_partition_key_ranges_builder::GetPartitionKeyRangesBuilder;
 pub use get_permission_builer::GetPermissionBuilder;
 pub use list_attachments_builder::ListAttachmentsBuilder;
 pub use list_collections_builder::ListCollectionsBuilder;
-pub use list_databases_builder::ListDatabasesBuilder;
 pub use list_documents_builder::ListDocumentsBuilder;
 pub use list_permissions_builder::ListPermissionsBuilder;
 pub use list_stored_procedures_builder::ListStoredProceduresBuilder;