diff --git a/sdk/data_cosmos/examples/collection.rs b/sdk/data_cosmos/examples/collection.rs
index bf1fd86589..50a36dce39 100644
--- a/sdk/data_cosmos/examples/collection.rs
+++ b/sdk/data_cosmos/examples/collection.rs
@@ -1,4 +1,3 @@
-use azure_core::prelude::*;
 use azure_data_cosmos::prelude::*;
 use futures::stream::StreamExt;
 use std::error::Error;
@@ -60,12 +59,10 @@ async fn main() -> Result<(), Box<dyn Error + Send + Sync>> {
 
     for db in databases.databases {
         let database_client = client.clone().into_database_client(db.id.clone());
-        let collections = Box::pin(
-            database_client.list_collections(Context::new(), ListCollectionsOptions::new()),
-        )
-        .next()
-        .await
-        .unwrap()?;
+        let collections = Box::pin(database_client.list_collections().into_stream())
+            .next()
+            .await
+            .unwrap()?;
         println!(
             "database {} has {} collection(s)",
             db.id,
diff --git a/sdk/data_cosmos/examples/create_delete_database.rs b/sdk/data_cosmos/examples/create_delete_database.rs
index 04be1b78b7..8d7f8fbf6b 100644
--- a/sdk/data_cosmos/examples/create_delete_database.rs
+++ b/sdk/data_cosmos/examples/create_delete_database.rs
@@ -1,4 +1,3 @@
-use azure_core::Context;
 use azure_data_cosmos::prelude::*;
 use futures::stream::StreamExt;
 use std::error::Error;
@@ -61,7 +60,7 @@ async fn main() -> Result<(), Box<dyn Error + Send + Sync>> {
         let get_collection_response = db_collection.get_collection().into_future().await?;
         println!("get_collection_response == {:#?}", get_collection_response);
 
-        let stream = db_client.list_collections(Context::new(), ListCollectionsOptions::new());
+        let stream = db_client.list_collections().into_stream();
         let mut stream = Box::pin(stream);
         while let Some(res) = stream.next().await {
             let res = res?;
diff --git a/sdk/data_cosmos/examples/database_00.rs b/sdk/data_cosmos/examples/database_00.rs
index 5f0d0c41cb..3fbc73c943 100644
--- a/sdk/data_cosmos/examples/database_00.rs
+++ b/sdk/data_cosmos/examples/database_00.rs
@@ -1,4 +1,3 @@
-use azure_core::Context;
 use azure_data_cosmos::prelude::*;
 use futures::stream::StreamExt;
 use serde_json::Value;
@@ -25,11 +24,10 @@ async fn main() -> Result<(), Box<dyn Error + Send + Sync>> {
         println!("database == {:?}", db);
         let database = client.clone().into_database_client(db.name().to_owned());
 
-        let collections =
-            Box::pin(database.list_collections(Context::new(), ListCollectionsOptions::new()))
-                .next()
-                .await
-                .unwrap()?;
+        let collections = Box::pin(database.list_collections().into_stream())
+            .next()
+            .await
+            .unwrap()?;
         for collection in collections.collections {
             println!("collection == {:?}", collection);
             let collection_client = database.clone().into_collection_client(collection.id);
@@ -66,10 +64,9 @@ async fn main() -> Result<(), Box<dyn Error + Send + Sync>> {
 
                 println!("\nReplacing collection");
                 let replace_collection_response = collection_client
-                    .replace_collection(
-                        Context::new(),
-                        ReplaceCollectionOptions::new("/age").indexing_policy(indexing_policy_new),
-                    )
+                    .replace_collection("/age")
+                    .indexing_policy(indexing_policy_new)
+                    .into_future()
                     .await?;
                 println!(
                     "replace_collection_response == {:#?}",
diff --git a/sdk/data_cosmos/examples/database_01.rs b/sdk/data_cosmos/examples/database_01.rs
index 878aa1f5c2..d0eaa3a034 100644
--- a/sdk/data_cosmos/examples/database_01.rs
+++ b/sdk/data_cosmos/examples/database_01.rs
@@ -1,4 +1,3 @@
-use azure_core::Context;
 use azure_data_cosmos::prelude::*;
 use futures::stream::StreamExt;
 use std::error::Error;
@@ -18,11 +17,10 @@ async fn main() -> Result<(), Box<dyn Error + Send + Sync>> {
     let database_client = client.into_database_client("pollo");
     println!("database_name == {}", database_client.database_name());
 
-    let collections =
-        Box::pin(database_client.list_collections(Context::new(), ListCollectionsOptions::new()))
-            .next()
-            .await
-            .unwrap()?;
+    let collections = Box::pin(database_client.list_collections().into_stream())
+        .next()
+        .await
+        .unwrap()?;
     println!("collections == {:#?}", collections);
 
     let collection_client = database_client.into_collection_client("cnt");
diff --git a/sdk/data_cosmos/examples/document_00.rs b/sdk/data_cosmos/examples/document_00.rs
index 842e15b740..2b145384d0 100644
--- a/sdk/data_cosmos/examples/document_00.rs
+++ b/sdk/data_cosmos/examples/document_00.rs
@@ -85,7 +85,8 @@ async fn main() -> Result<(), Box<dyn Error + Send + Sync>> {
             client
                 .clone()
                 .into_database_client(database.id.clone())
-                .list_collections(Context::new(), ListCollectionsOptions::new()),
+                .list_collections()
+                .into_stream(),
         )
         .next()
         .await
@@ -156,7 +157,8 @@ async fn main() -> Result<(), Box<dyn Error + Send + Sync>> {
     let get_document_response = collection_client
         .clone()
         .into_document_client(doc.id.clone(), &doc.id)?
-        .get_document::<MySampleStruct>(Context::new(), GetDocumentOptions::new())
+        .get_document()
+        .into_future::<MySampleStruct>()
         .await?;
     println!("get_document_response == {:#?}", get_document_response);
 
@@ -170,12 +172,12 @@ async fn main() -> Result<(), Box<dyn Error + Send + Sync>> {
         // the etag received in the previous get_document. The etag is an opaque value that
         // changes every time the document is updated. If the passed etag is different in
         // CosmosDB it means something else updated the document before us!
-        let options = ReplaceDocumentOptions::new()
-            .if_match_condition(IfMatchCondition::Match(document.etag));
         let replace_document_response = collection_client
             .clone()
             .into_document_client(doc.id.clone(), &doc.id)?
-            .replace_document(Context::new(), &doc, options)
+            .replace_document(doc)
+            .if_match_condition(IfMatchCondition::Match(document.etag))
+            .into_future()
             .await?;
         println!(
             "replace_document_response == {:#?}",
diff --git a/sdk/data_cosmos/examples/document_entries_00.rs b/sdk/data_cosmos/examples/document_entries_00.rs
index 8c393ed6b8..d4b4063afb 100644
--- a/sdk/data_cosmos/examples/document_entries_00.rs
+++ b/sdk/data_cosmos/examples/document_entries_00.rs
@@ -115,13 +115,11 @@ async fn main() -> Result<(), Box<dyn Error + Send + Sync>> {
     let id = format!("unique_id{}", 3);
     let partition_key = &id;
 
-    let response = client
+    let response: GetDocumentResponse<MySampleStruct> = client
         .clone()
         .into_document_client(id.clone(), partition_key)?
-        .get_document::<MySampleStruct>(
-            Context::new(),
-            GetDocumentOptions::new().consistency_level(session_token),
-        )
+        .get_document()
+        .into_future()
         .await?;
 
     assert!(matches!(response, GetDocumentResponse::Found(_)));
@@ -137,13 +135,10 @@ async fn main() -> Result<(), Box<dyn Error + Send + Sync>> {
     let replace_document_response = client
         .clone()
         .into_document_client(id.clone(), &id)?
-        .replace_document(
-            Context::new(),
-            &doc.document,
-            ReplaceDocumentOptions::new()
-                .consistency_level(ConsistencyLevel::from(&response))
-                .if_match_condition(IfMatchCondition::Match(doc.etag)), // use optimistic concurrency check
-        )
+        .replace_document(doc.document)
+        .consistency_level(ConsistencyLevel::from(&response))
+        .if_match_condition(IfMatchCondition::Match(doc.etag)) // use optimistic concurrency check
+        .into_future()
         .await?;
 
     println!(
@@ -155,14 +150,12 @@ async fn main() -> Result<(), Box<dyn Error + Send + Sync>> {
     // has_been_found == false
     println!("\n\nLooking for non-existing item");
     let id = format!("unique_id{}", 100);
-
-    let response = client
+    let response: GetDocumentResponse<MySampleStruct> = client
         .clone()
         .into_document_client(id.clone(), &id)?
-        .get_document::<MySampleStruct>(
-            Context::new(),
-            GetDocumentOptions::new().consistency_level(&response),
-        )
+        .get_document()
+        .consistency_level(&response)
+        .into_future()
         .await?;
 
     assert!(matches!(response, GetDocumentResponse::NotFound(_)));
diff --git a/sdk/data_cosmos/examples/document_entries_01.rs b/sdk/data_cosmos/examples/document_entries_01.rs
index 50beb3c505..74aadefc96 100644
--- a/sdk/data_cosmos/examples/document_entries_01.rs
+++ b/sdk/data_cosmos/examples/document_entries_01.rs
@@ -1,4 +1,3 @@
-use azure_core::Context;
 use azure_data_cosmos::prelude::*;
 use serde::{Deserialize, Serialize};
 use std::error::Error;
@@ -62,20 +61,18 @@ async fn main() -> Result<(), Box<dyn Error + Send + Sync>> {
     let get_document_response = client
         .clone()
         .into_document_client(doc.id.clone(), &doc.id)?
-        .get_document::<serde_json::Value>(
-            Context::new(),
-            GetDocumentOptions::new().consistency_level(&create_document_response),
-        )
+        .get_document()
+        .consistency_level(&create_document_response)
+        .into_future::<serde_json::Value>()
         .await?;
     println!("get_document_response == {:#?}", get_document_response);
 
     let get_document_response = client
         .clone()
         .into_document_client("ciccia", &doc.id)?
-        .get_document::<serde_json::Value>(
-            Context::new(),
-            GetDocumentOptions::new().consistency_level(&create_document_response),
-        )
+        .get_document()
+        .consistency_level(&create_document_response)
+        .into_future::<serde_json::Value>()
         .await?;
     println!(
         "get_document_response == {:#?}\n\n\n",
@@ -104,11 +101,9 @@ async fn main() -> Result<(), Box<dyn Error + Send + Sync>> {
 
     let replace_document_response = client
         .into_document_client(doc.id.clone(), &doc.id)?
-        .replace_document(
-            Context::new(),
-            &doc,
-            ReplaceDocumentOptions::new().consistency_level(&query_documents_response),
-        )
+        .replace_document(doc)
+        .consistency_level(&query_documents_response)
+        .into_future()
         .await?;
     println!(
         "replace_document_response == {:#?}",
diff --git a/sdk/data_cosmos/examples/permission_00.rs b/sdk/data_cosmos/examples/permission_00.rs
index 9dd77a4fdc..9f1e696212 100644
--- a/sdk/data_cosmos/examples/permission_00.rs
+++ b/sdk/data_cosmos/examples/permission_00.rs
@@ -101,12 +101,11 @@ async fn main() -> Result<(), Box<dyn Error + Send + Sync>> {
     );
 
     let get_permission_response = permission_client
-        .get_permission(
-            Context::new(),
-            GetPermissionOptions::new().consistency_level(ConsistencyLevel::Session(
-                list_permissions_response.session_token,
-            )),
-        )
+        .get_permission()
+        .consistency_level(ConsistencyLevel::Session(
+            list_permissions_response.session_token,
+        ))
+        .into_future()
         .await
         .unwrap();
     println!("get_permission_response == {:#?}", get_permission_response);
diff --git a/sdk/data_cosmos/examples/user_00.rs b/sdk/data_cosmos/examples/user_00.rs
index cc14770279..5b065b870f 100644
--- a/sdk/data_cosmos/examples/user_00.rs
+++ b/sdk/data_cosmos/examples/user_00.rs
@@ -1,6 +1,6 @@
 use azure_core::Context;
 use azure_data_cosmos::prelude::*;
-use futures::stream::StreamExt;
+use futures::StreamExt;
 use std::error::Error;
 
 #[tokio::main]
@@ -32,16 +32,14 @@ async fn main() -> Result<(), Box<dyn Error + Send + Sync>> {
     let create_user_response = user_client.create_user().into_future().await?;
     println!("create_user_response == {:#?}", create_user_response);
 
-    let users = Box::pin(database_client.list_users(Context::new(), ListUsersOptions::new()))
+    let users = Box::pin(database_client.list_users().into_stream())
         .next()
         .await
         .unwrap()?;
 
     println!("list_users_response == {:#?}", users);
 
-    let get_user_response = user_client
-        .get_user(Context::new(), GetUserOptions::new())
-        .await?;
+    let get_user_response = user_client.get_user().into_future().await?;
     println!("get_user_response == {:#?}", get_user_response);
 
     let new_user = format!("{}replaced", user_name);
diff --git a/sdk/data_cosmos/src/clients/collection_client.rs b/sdk/data_cosmos/src/clients/collection_client.rs
index add52fee4b..61ecdfe792 100644
--- a/sdk/data_cosmos/src/clients/collection_client.rs
+++ b/sdk/data_cosmos/src/clients/collection_client.rs
@@ -2,10 +2,10 @@ use super::{DatabaseClient, UserDefinedFunctionClient};
 use crate::clients::*;
 use crate::operations::*;
 use crate::requests;
-use crate::resources::ResourceType;
+use crate::resources::collection::PartitionKey;
 use crate::CosmosEntity;
 use crate::ReadonlyString;
-use azure_core::{Context, HttpClient, Pipeline, Request};
+use azure_core::{HttpClient, Pipeline, Request};
 use serde::Serialize;
 
 /// A client for Cosmos collection resources.
@@ -52,21 +52,11 @@ impl CollectionClient {
     }
 
     /// Replace a collection
-    pub async fn replace_collection(
+    pub fn replace_collection<P: Into<PartitionKey>>(
         &self,
-        ctx: Context,
-        options: ReplaceCollectionOptions,
-    ) -> crate::Result<ReplaceCollectionResponse> {
-        let mut request = self.prepare_request_with_collection_name(http::Method::PUT);
-
-        options.decorate_request(&mut request, self.collection_name())?;
-
-        let response = self
-            .pipeline()
-            .send(ctx.clone().insert(ResourceType::Collections), &mut request)
-            .await?;
-
-        Ok(ReplaceCollectionResponse::try_from(response).await?)
+        partition_key: P,
+    ) -> ReplaceCollectionBuilder {
+        ReplaceCollectionBuilder::new(self.clone(), partition_key.into())
     }
 
     /// list documents in a collection
diff --git a/sdk/data_cosmos/src/clients/cosmos_client.rs b/sdk/data_cosmos/src/clients/cosmos_client.rs
index fdd4ee141d..db28051e61 100644
--- a/sdk/data_cosmos/src/clients/cosmos_client.rs
+++ b/sdk/data_cosmos/src/clients/cosmos_client.rs
@@ -168,8 +168,8 @@ impl CosmosClient {
     }
 
     /// List all databases
-    pub fn list_databases(&self) -> ListDatabases {
-        ListDatabases::new(self.clone())
+    pub fn list_databases(&self) -> ListDatabasesBuilder {
+        ListDatabasesBuilder::new(self.clone())
     }
 
     /// Convert into a [`DatabaseClient`]
diff --git a/sdk/data_cosmos/src/clients/database_client.rs b/sdk/data_cosmos/src/clients/database_client.rs
index 4220f7a3c2..ad921c6c1f 100644
--- a/sdk/data_cosmos/src/clients/database_client.rs
+++ b/sdk/data_cosmos/src/clients/database_client.rs
@@ -1,33 +1,8 @@
 use super::*;
 use crate::operations::*;
 use crate::resources::collection::PartitionKey;
-use crate::resources::ResourceType;
 use crate::ReadonlyString;
-use azure_core::prelude::Continuation;
 use azure_core::Pipeline;
-use azure_core::{AddAsHeader, Context};
-use futures::stream::unfold;
-use futures::Stream;
-
-/// Macro for short cutting a stream on error
-macro_rules! r#try {
-    ($expr:expr $(,)?) => {
-        match $expr {
-            Result::Ok(val) => val,
-            Result::Err(err) => {
-                return Some((Err(err.into()), State::Done));
-            }
-        }
-    };
-}
-
-/// Stream state
-#[derive(Debug, Clone, PartialEq)]
-enum State {
-    Init,
-    Continuation(String),
-    Done,
-}
 
 /// A client for Cosmos database resources.
 #[derive(Debug, Clone)]
@@ -68,61 +43,8 @@ impl DatabaseClient {
     }
 
     /// List collections in the database
-    pub fn list_collections(
-        &self,
-        ctx: Context,
-        options: ListCollectionsOptions,
-    ) -> impl Stream<Item = crate::Result<ListCollectionsResponse>> + '_ {
-        unfold(State::Init, move |state: State| {
-            let this = self.clone();
-            let ctx = ctx.clone();
-            let options = options.clone();
-            async move {
-                let response = match state {
-                    State::Init => {
-                        let mut request = this.cosmos_client().prepare_request_pipeline(
-                            &format!("dbs/{}/colls", this.database_name()),
-                            http::Method::GET,
-                        );
-
-                        r#try!(options.decorate_request(&mut request));
-                        let response = r#try!(
-                            this.pipeline()
-                                .send(ctx.clone().insert(ResourceType::Collections), &mut request)
-                                .await
-                        );
-                        ListCollectionsResponse::try_from(response).await
-                    }
-                    State::Continuation(continuation_token) => {
-                        let continuation = Continuation::new(continuation_token.as_str());
-                        let mut request = this.cosmos_client().prepare_request_pipeline(
-                            &format!("dbs/{}/colls", self.database_name()),
-                            http::Method::GET,
-                        );
-
-                        r#try!(options.decorate_request(&mut request));
-                        r#try!(continuation.add_as_header2(&mut request));
-                        let response = r#try!(
-                            this.pipeline()
-                                .send(ctx.clone().insert(ResourceType::Collections), &mut request)
-                                .await
-                        );
-                        ListCollectionsResponse::try_from(response).await
-                    }
-                    State::Done => return None,
-                };
-
-                let response = r#try!(response);
-
-                let next_state = response
-                    .continuation_token
-                    .clone()
-                    .map(State::Continuation)
-                    .unwrap_or(State::Done);
-
-                Some((Ok(response), next_state))
-            }
-        })
+    pub fn list_collections(&self) -> ListCollectionsBuilder {
+        ListCollectionsBuilder::new(self.clone())
     }
 
     /// Create a collection
@@ -135,61 +57,8 @@ impl DatabaseClient {
     }
 
     /// List users
-    pub fn list_users(
-        &self,
-        ctx: Context,
-        options: ListUsersOptions,
-    ) -> impl Stream<Item = crate::Result<ListUsersResponse>> + '_ {
-        unfold(State::Init, move |state: State| {
-            let this = self.clone();
-            let ctx = ctx.clone();
-            let options = options.clone();
-            async move {
-                let response = match state {
-                    State::Init => {
-                        let mut request = this.cosmos_client().prepare_request_pipeline(
-                            &format!("dbs/{}/users", this.database_name()),
-                            http::Method::GET,
-                        );
-
-                        r#try!(options.decorate_request(&mut request));
-                        let response = r#try!(
-                            this.pipeline()
-                                .send(ctx.clone().insert(ResourceType::Users), &mut request)
-                                .await
-                        );
-                        ListUsersResponse::try_from(response).await
-                    }
-                    State::Continuation(continuation_token) => {
-                        let continuation = Continuation::new(continuation_token.as_str());
-                        let mut request = this.cosmos_client().prepare_request_pipeline(
-                            &format!("dbs/{}/users", self.database_name()),
-                            http::Method::GET,
-                        );
-
-                        r#try!(options.decorate_request(&mut request));
-                        r#try!(continuation.add_as_header2(&mut request));
-                        let response = r#try!(
-                            this.pipeline()
-                                .send(ctx.clone().insert(ResourceType::Users), &mut request)
-                                .await
-                        );
-                        ListUsersResponse::try_from(response).await
-                    }
-                    State::Done => return None,
-                };
-
-                let response = r#try!(response);
-
-                let next_state = response
-                    .continuation_token
-                    .clone()
-                    .map(State::Continuation)
-                    .unwrap_or_else(|| State::Done);
-
-                Some((Ok(response), next_state))
-            }
-        })
+    pub fn list_users(&self) -> ListUsersBuilder {
+        ListUsersBuilder::new(self.clone())
     }
 
     /// Convert into a [`CollectionClient`]
diff --git a/sdk/data_cosmos/src/clients/document_client.rs b/sdk/data_cosmos/src/clients/document_client.rs
index 7a9ff24b64..836b6618ba 100644
--- a/sdk/data_cosmos/src/clients/document_client.rs
+++ b/sdk/data_cosmos/src/clients/document_client.rs
@@ -1,9 +1,7 @@
 use super::{AttachmentClient, CollectionClient, CosmosClient, DatabaseClient};
 use crate::operations::*;
-use crate::resources::ResourceType;
 use crate::{requests, ReadonlyString};
-use azure_core::{Context, HttpClient, Request};
-use serde::de::DeserializeOwned;
+use azure_core::{HttpClient, Request};
 use serde::Serialize;
 
 /// A client for Cosmos document resources.
@@ -59,45 +57,16 @@ impl DocumentClient {
     }
 
     /// Get a document
-    pub async fn get_document<T>(
-        &self,
-        ctx: Context,
-        options: GetDocumentOptions,
-    ) -> crate::Result<GetDocumentResponse<T>>
-    where
-        T: DeserializeOwned,
-    {
-        let mut request = self.prepare_request_pipeline_with_document_name(http::Method::GET);
-
-        options.decorate_request(&mut request)?;
-
-        let response = self
-            .cosmos_client()
-            .pipeline()
-            .send(ctx.clone().insert(ResourceType::Documents), &mut request)
-            .await?;
-
-        GetDocumentResponse::try_from(response).await
+    pub fn get_document(&self) -> GetDocumentBuilder {
+        GetDocumentBuilder::new(self.clone())
     }
 
     /// replace a document in a collection
-    pub async fn replace_document<T: Serialize>(
+    pub fn replace_document<D: Serialize + Send + 'static>(
         &self,
-        ctx: Context,
-        document: &T,
-        options: ReplaceDocumentOptions,
-    ) -> crate::Result<ReplaceDocumentResponse> {
-        let mut request = self.prepare_request_pipeline_with_document_name(http::Method::PUT);
-
-        options.decorate_request(&mut request, document, self.partition_key_serialized())?;
-
-        let response = self
-            .cosmos_client()
-            .pipeline()
-            .send(ctx.clone().insert(ResourceType::Documents), &mut request)
-            .await?;
-
-        ReplaceDocumentResponse::try_from(response).await
+        document: D,
+    ) -> ReplaceDocumentBuilder<D> {
+        ReplaceDocumentBuilder::new(self.clone(), document)
     }
 
     /// Delete a document
diff --git a/sdk/data_cosmos/src/clients/permission_client.rs b/sdk/data_cosmos/src/clients/permission_client.rs
index a8adaf4afc..f317c89a1c 100644
--- a/sdk/data_cosmos/src/clients/permission_client.rs
+++ b/sdk/data_cosmos/src/clients/permission_client.rs
@@ -68,21 +68,8 @@ impl PermissionClient {
     }
 
     /// Get the permission
-    pub async fn get_permission(
-        &self,
-        ctx: Context,
-        options: GetPermissionOptions,
-    ) -> crate::Result<PermissionResponse> {
-        let mut request = self.prepare_request_with_permission_name(http::Method::GET);
-
-        options.decorate_request(&mut request)?;
-
-        let response = self
-            .pipeline()
-            .send(ctx.clone().insert(ResourceType::Permissions), &mut request)
-            .await?;
-
-        Ok(PermissionResponse::try_from(response).await?)
+    pub fn get_permission(&self) -> GetPermissionBuilder {
+        GetPermissionBuilder::new(self.clone())
     }
 
     /// Delete the permission
diff --git a/sdk/data_cosmos/src/clients/user_client.rs b/sdk/data_cosmos/src/clients/user_client.rs
index fc385640ab..6cb1ca45c5 100644
--- a/sdk/data_cosmos/src/clients/user_client.rs
+++ b/sdk/data_cosmos/src/clients/user_client.rs
@@ -44,20 +44,8 @@ impl UserClient {
     }
 
     /// Get the user
-    pub async fn get_user(
-        &self,
-        ctx: Context,
-        options: GetUserOptions,
-    ) -> crate::Result<UserResponse> {
-        let mut request = self.prepare_request_with_user_name(http::Method::GET);
-
-        options.decorate_request(&mut request)?;
-        let response = self
-            .pipeline()
-            .send(ctx.clone().insert(ResourceType::Users), &mut request)
-            .await?;
-
-        Ok(UserResponse::try_from(response).await?)
+    pub fn get_user(&self) -> GetUserBuilder {
+        GetUserBuilder::new(self.clone())
     }
 
     /// Replace the user
diff --git a/sdk/data_cosmos/src/operations/create_collection.rs b/sdk/data_cosmos/src/operations/create_collection.rs
index c4a02de6b2..930207732b 100644
--- a/sdk/data_cosmos/src/operations/create_collection.rs
+++ b/sdk/data_cosmos/src/operations/create_collection.rs
@@ -80,7 +80,8 @@ impl std::future::IntoFuture for CreateCollectionBuilder {
     }
 }
 
-type CreateCollection =
+/// The future returned by calling `into_future` on the builder.
+pub type CreateCollection =
     futures::future::BoxFuture<'static, crate::Result<CreateCollectionResponse>>;
 
 /// Body for the create collection request
diff --git a/sdk/data_cosmos/src/operations/create_database.rs b/sdk/data_cosmos/src/operations/create_database.rs
index 31f06b0157..cb8f7423a0 100644
--- a/sdk/data_cosmos/src/operations/create_database.rs
+++ b/sdk/data_cosmos/src/operations/create_database.rs
@@ -73,7 +73,8 @@ impl std::future::IntoFuture for CreateDatabaseBuilder {
     }
 }
 
-type CreateDatabase =
+/// The future returned by calling `into_future` on the builder.
+pub type CreateDatabase =
     futures::future::BoxFuture<'static, azure_core::error::Result<CreateDatabaseResponse>>;
 
 #[derive(Serialize)]
diff --git a/sdk/data_cosmos/src/operations/create_document.rs b/sdk/data_cosmos/src/operations/create_document.rs
index ca315c188b..380f5c889c 100644
--- a/sdk/data_cosmos/src/operations/create_document.rs
+++ b/sdk/data_cosmos/src/operations/create_document.rs
@@ -93,7 +93,9 @@ impl<D: Serialize + CosmosEntity + Send + 'static> CreateDocumentBuilder<D> {
     }
 }
 
-type CreateDocument = futures::future::BoxFuture<'static, crate::Result<CreateDocumentResponse>>;
+/// The future returned by calling `into_future` on the builder.
+pub type CreateDocument =
+    futures::future::BoxFuture<'static, crate::Result<CreateDocumentResponse>>;
 
 #[cfg(feature = "into_future")]
 impl<D: Serialize + CosmosEntity + Send + 'static> std::future::IntoFuture
diff --git a/sdk/data_cosmos/src/operations/create_permission.rs b/sdk/data_cosmos/src/operations/create_permission.rs
index ff80e3547e..743b8c2d12 100644
--- a/sdk/data_cosmos/src/operations/create_permission.rs
+++ b/sdk/data_cosmos/src/operations/create_permission.rs
@@ -72,7 +72,8 @@ impl CreatePermissionBuilder {
     }
 }
 
-type CreatePermission = futures::future::BoxFuture<'static, crate::Result<PermissionResponse>>;
+/// The future returned by calling `into_future` on the builder.
+pub type CreatePermission = futures::future::BoxFuture<'static, crate::Result<PermissionResponse>>;
 
 #[cfg(feature = "into_future")]
 impl std::future::IntoFuture for CreatePermissionBuilder {
diff --git a/sdk/data_cosmos/src/operations/create_user.rs b/sdk/data_cosmos/src/operations/create_user.rs
index 0e16bd8fab..4909710e2e 100644
--- a/sdk/data_cosmos/src/operations/create_user.rs
+++ b/sdk/data_cosmos/src/operations/create_user.rs
@@ -51,7 +51,8 @@ impl CreateUserBuilder {
     }
 }
 
-type CreateUser = futures::future::BoxFuture<'static, crate::Result<UserResponse>>;
+/// The future returned by calling `into_future` on the builder.
+pub type CreateUser = futures::future::BoxFuture<'static, crate::Result<UserResponse>>;
 
 #[cfg(feature = "into_future")]
 impl std::future::IntoFuture for CreateUserBuilder {
diff --git a/sdk/data_cosmos/src/operations/delete_collection.rs b/sdk/data_cosmos/src/operations/delete_collection.rs
index df1e53c93e..1a52fc1bb1 100644
--- a/sdk/data_cosmos/src/operations/delete_collection.rs
+++ b/sdk/data_cosmos/src/operations/delete_collection.rs
@@ -47,7 +47,8 @@ impl DeleteCollectionBuilder {
     }
 }
 
-type DeleteCollection =
+/// The future returned by calling `into_future` on the builder.
+pub type DeleteCollection =
     futures::future::BoxFuture<'static, crate::Result<DeleteCollectionResponse>>;
 
 #[cfg(feature = "into_future")]
diff --git a/sdk/data_cosmos/src/operations/delete_database.rs b/sdk/data_cosmos/src/operations/delete_database.rs
index ba3cf58670..5e7fd11ce7 100644
--- a/sdk/data_cosmos/src/operations/delete_database.rs
+++ b/sdk/data_cosmos/src/operations/delete_database.rs
@@ -48,7 +48,9 @@ impl DeleteDatabaseBuilder {
     }
 }
 
-type DeleteDatabase = futures::future::BoxFuture<'static, crate::Result<DeleteDatabaseResponse>>;
+/// The future returned by calling `into_future` on the builder.
+pub type DeleteDatabase =
+    futures::future::BoxFuture<'static, crate::Result<DeleteDatabaseResponse>>;
 
 #[cfg(feature = "into_future")]
 impl std::future::IntoFuture for DeleteDatabaseBuilder {
diff --git a/sdk/data_cosmos/src/operations/delete_document.rs b/sdk/data_cosmos/src/operations/delete_document.rs
index 9078a63298..26691c2abb 100644
--- a/sdk/data_cosmos/src/operations/delete_document.rs
+++ b/sdk/data_cosmos/src/operations/delete_document.rs
@@ -67,7 +67,9 @@ impl DeleteDocumentBuilder {
     }
 }
 
-type DeleteDocument = futures::future::BoxFuture<'static, crate::Result<DeleteDocumentResponse>>;
+/// The future returned by calling `into_future` on the builder.
+pub type DeleteDocument =
+    futures::future::BoxFuture<'static, crate::Result<DeleteDocumentResponse>>;
 
 #[cfg(feature = "into_future")]
 impl std::future::IntoFuture for DeleteDocumentBuilder {
diff --git a/sdk/data_cosmos/src/operations/delete_permission.rs b/sdk/data_cosmos/src/operations/delete_permission.rs
index a40a9f9caf..971d7b0e0c 100644
--- a/sdk/data_cosmos/src/operations/delete_permission.rs
+++ b/sdk/data_cosmos/src/operations/delete_permission.rs
@@ -48,7 +48,8 @@ impl DeletePermissionBuilder {
     }
 }
 
-type DeletePermission =
+/// The future returned by calling `into_future` on the builder.
+pub type DeletePermission =
     futures::future::BoxFuture<'static, crate::Result<DeletePermissionResponse>>;
 
 #[cfg(feature = "into_future")]
diff --git a/sdk/data_cosmos/src/operations/delete_user.rs b/sdk/data_cosmos/src/operations/delete_user.rs
index 010c5ce176..d6371f9077 100644
--- a/sdk/data_cosmos/src/operations/delete_user.rs
+++ b/sdk/data_cosmos/src/operations/delete_user.rs
@@ -44,7 +44,8 @@ impl DeleteUserBuilder {
     }
 }
 
-type DeleteUser = futures::future::BoxFuture<'static, crate::Result<DeleteUserResponse>>;
+/// The future returned by calling `into_future` on the builder.
+pub type DeleteUser = futures::future::BoxFuture<'static, crate::Result<DeleteUserResponse>>;
 
 #[cfg(feature = "into_future")]
 impl std::future::IntoFuture for DeleteUserBuilder {
diff --git a/sdk/data_cosmos/src/operations/get_collection.rs b/sdk/data_cosmos/src/operations/get_collection.rs
index 14127119c6..6707090e8a 100644
--- a/sdk/data_cosmos/src/operations/get_collection.rs
+++ b/sdk/data_cosmos/src/operations/get_collection.rs
@@ -50,7 +50,8 @@ impl GetCollectionBuilder {
     }
 }
 
-type GetCollection = futures::future::BoxFuture<'static, crate::Result<GetCollectionResponse>>;
+/// The future returned by calling `into_future` on the builder.
+pub type GetCollection = futures::future::BoxFuture<'static, crate::Result<GetCollectionResponse>>;
 
 #[cfg(feature = "into_future")]
 impl std::future::IntoFuture for GetCollectionBuilder {
diff --git a/sdk/data_cosmos/src/operations/get_database.rs b/sdk/data_cosmos/src/operations/get_database.rs
index d1904e2fac..d211f780ac 100644
--- a/sdk/data_cosmos/src/operations/get_database.rs
+++ b/sdk/data_cosmos/src/operations/get_database.rs
@@ -51,7 +51,8 @@ impl GetDatabaseBuilder {
     }
 }
 
-type GetDatabase = futures::future::BoxFuture<'static, crate::Result<GetDatabaseResponse>>;
+/// The future returned by calling `into_future` on the builder.
+pub type GetDatabase = futures::future::BoxFuture<'static, crate::Result<GetDatabaseResponse>>;
 
 #[cfg(feature = "into_future")]
 impl std::future::IntoFuture for GetDatabaseBuilder {
diff --git a/sdk/data_cosmos/src/operations/get_document.rs b/sdk/data_cosmos/src/operations/get_document.rs
index 977a2e81b5..49f6b23b38 100644
--- a/sdk/data_cosmos/src/operations/get_document.rs
+++ b/sdk/data_cosmos/src/operations/get_document.rs
@@ -4,26 +4,28 @@ use crate::resources::Document;
 use crate::ResourceQuota;
 use azure_core::headers::{etag_from_headers, session_token_from_headers};
 use azure_core::prelude::*;
-use azure_core::{
-    collect_pinned_stream, Request as HttpRequest, Response as HttpResponse, SessionToken,
-};
+use azure_core::{collect_pinned_stream, Response as HttpResponse, SessionToken};
 use chrono::{DateTime, Utc};
 use http::{HeaderMap, StatusCode};
 use serde::de::DeserializeOwned;
 
 #[derive(Debug, Clone)]
-pub struct GetDocumentOptions {
+pub struct GetDocumentBuilder {
+    client: DocumentClient,
     if_match_condition: Option<IfMatchCondition>,
     if_modified_since: Option<IfModifiedSince>,
     consistency_level: Option<ConsistencyLevel>,
+    context: Context,
 }
 
-impl GetDocumentOptions {
-    pub fn new() -> Self {
+impl GetDocumentBuilder {
+    pub(crate) fn new(client: DocumentClient) -> Self {
         Self {
+            client,
             if_match_condition: None,
             if_modified_since: None,
             consistency_level: None,
+            context: Context::new(),
         }
     }
 
@@ -31,23 +33,53 @@ impl GetDocumentOptions {
         consistency_level: ConsistencyLevel => Some(consistency_level),
         if_match_condition: IfMatchCondition => Some(if_match_condition),
         if_modified_since: DateTime<Utc> => Some(IfModifiedSince::new(if_modified_since)),
+        context: Context => context,
     }
 
-    pub(crate) fn decorate_request(&self, request: &mut HttpRequest) -> crate::Result<()> {
-        azure_core::headers::add_optional_header2(&self.if_match_condition, request)?;
-        azure_core::headers::add_optional_header2(&self.if_modified_since, request)?;
-        azure_core::headers::add_optional_header2(&self.consistency_level, request)?;
-
-        request.set_body(azure_core::EMPTY_BODY.into());
-
-        Ok(())
+    /// Convert into a future
+    ///
+    /// We do not implement `std::future::IntoFuture` because it requires the ability for the
+    /// output of the future to be generic which is not possible in Rust (as of 1.59). Once
+    /// generic associated types (GATs) stabilize, this will become possible.
+    pub fn into_future<T: DeserializeOwned>(self) -> GetDocument<T> {
+        Box::pin(async move {
+            let mut request = self
+                .client
+                .prepare_request_pipeline_with_document_name(http::Method::GET);
+
+            azure_core::headers::add_optional_header2(&self.if_match_condition, &mut request)?;
+            azure_core::headers::add_optional_header2(&self.if_modified_since, &mut request)?;
+            azure_core::headers::add_optional_header2(&self.consistency_level, &mut request)?;
+
+            request.set_body(azure_core::EMPTY_BODY.into());
+
+            let response = self
+                .client
+                .cosmos_client()
+                .pipeline()
+                .send(
+                    self.context.clone().insert(ResourceType::Documents),
+                    &mut request,
+                )
+                .await?;
+
+            GetDocumentResponse::try_from(response).await
+        })
     }
 }
 
+/// The future returned by calling `into_future` on the builder.
+pub type GetDocument<T> =
+    futures::future::BoxFuture<'static, crate::Result<GetDocumentResponse<T>>>;
+
 #[derive(Debug, Clone)]
+// note(rylev): clippy seems to be falsely detecting that
+// one of the variants is much larger than the other (which
+// is not true)
+#[allow(clippy::large_enum_variant)]
 pub enum GetDocumentResponse<T> {
-    Found(Box<FoundDocumentResponse<T>>),
-    NotFound(Box<NotFoundDocumentResponse>),
+    Found(FoundDocumentResponse<T>),
+    NotFound(NotFoundDocumentResponse),
 }
 
 impl<T> GetDocumentResponse<T>
@@ -63,13 +95,13 @@ where
         let body = collect_pinned_stream(pinned_stream).await?;
 
         if has_been_found {
-            Ok(GetDocumentResponse::Found(Box::new(
+            Ok(GetDocumentResponse::Found(
                 FoundDocumentResponse::try_from(&headers, body).await?,
-            )))
+            ))
         } else {
-            Ok(GetDocumentResponse::NotFound(Box::new(
+            Ok(GetDocumentResponse::NotFound(
                 NotFoundDocumentResponse::try_from(&headers).await?,
-            )))
+            ))
         }
     }
 }
diff --git a/sdk/data_cosmos/src/operations/get_permission.rs b/sdk/data_cosmos/src/operations/get_permission.rs
index 223ae44352..891d1eeb94 100644
--- a/sdk/data_cosmos/src/operations/get_permission.rs
+++ b/sdk/data_cosmos/src/operations/get_permission.rs
@@ -1,26 +1,58 @@
-use crate::prelude::*;
+use crate::{prelude::*, resources::permission::PermissionResponse};
 
-use azure_core::Request as HttpRequest;
+use azure_core::Context;
 
 #[derive(Debug, Clone)]
-pub struct GetPermissionOptions {
+pub struct GetPermissionBuilder {
+    client: PermissionClient,
     consistency_level: Option<ConsistencyLevel>,
+    context: Context,
 }
 
-impl GetPermissionOptions {
-    pub fn new() -> Self {
+impl GetPermissionBuilder {
+    pub fn new(client: PermissionClient) -> Self {
         Self {
+            client,
             consistency_level: None,
+            context: Context::new(),
         }
     }
 
     setters! {
         consistency_level: ConsistencyLevel => Some(consistency_level),
+        context: Context => context,
     }
 
-    pub(crate) fn decorate_request(&self, request: &mut HttpRequest) -> crate::Result<()> {
-        azure_core::headers::add_optional_header2(&self.consistency_level, request)?;
+    pub fn into_future(self) -> GetPermission {
+        Box::pin(async move {
+            let mut request = self
+                .client
+                .prepare_request_with_permission_name(http::Method::GET);
 
-        Ok(())
+            azure_core::headers::add_optional_header2(&self.consistency_level, &mut request)?;
+
+            let response = self
+                .client
+                .pipeline()
+                .send(
+                    self.context.clone().insert(ResourceType::Permissions),
+                    &mut request,
+                )
+                .await?;
+
+            PermissionResponse::try_from(response).await
+        })
+    }
+}
+
+/// The future returned by calling `into_future` on the builder.
+pub type GetPermission = futures::future::BoxFuture<'static, crate::Result<PermissionResponse>>;
+
+#[cfg(feature = "into_future")]
+impl std::future::IntoFuture for GetPermissionBuilder {
+    type Future = GetPermission;
+    type Output = <GetPermission as std::future::Future>::Output;
+    fn into_future(self) -> Self::Future {
+        Self::into_future(self)
     }
 }
diff --git a/sdk/data_cosmos/src/operations/get_user.rs b/sdk/data_cosmos/src/operations/get_user.rs
index b6c8b694b3..4f8deb4077 100644
--- a/sdk/data_cosmos/src/operations/get_user.rs
+++ b/sdk/data_cosmos/src/operations/get_user.rs
@@ -1,26 +1,57 @@
-use crate::prelude::*;
-use azure_core::Request as HttpRequest;
+use crate::{prelude::*, resources::user::UserResponse};
+use azure_core::Context;
 
-#[derive(Debug, Clone, Default)]
-pub struct GetUserOptions {
+#[derive(Debug, Clone)]
+pub struct GetUserBuilder {
+    client: UserClient,
     consistency_level: Option<ConsistencyLevel>,
+    context: Context,
 }
 
-impl GetUserOptions {
-    pub fn new() -> Self {
+impl GetUserBuilder {
+    pub fn new(client: UserClient) -> Self {
         Self {
+            client,
             consistency_level: None,
+            context: Context::new(),
         }
     }
 
     setters! {
         consistency_level: ConsistencyLevel => Some(consistency_level),
+        context: Context => context,
     }
 
-    pub(crate) fn decorate_request(&self, request: &mut HttpRequest) -> crate::Result<()> {
-        azure_core::headers::add_optional_header2(&self.consistency_level, request)?;
-        request.set_body(bytes::Bytes::from_static(&[]).into());
+    pub fn into_future(self) -> GetUser {
+        Box::pin(async move {
+            let mut request = self
+                .client
+                .prepare_request_with_user_name(http::Method::GET);
 
-        Ok(())
+            azure_core::headers::add_optional_header2(&self.consistency_level, &mut request)?;
+            request.set_body(bytes::Bytes::from_static(&[]).into());
+            let response = self
+                .client
+                .pipeline()
+                .send(
+                    self.context.clone().insert(ResourceType::Users),
+                    &mut request,
+                )
+                .await?;
+
+            UserResponse::try_from(response).await
+        })
+    }
+}
+
+/// The future returned by calling `into_future` on the builder.
+pub type GetUser = futures::future::BoxFuture<'static, crate::Result<UserResponse>>;
+
+#[cfg(feature = "into_future")]
+impl std::future::IntoFuture for GetUserBuilder {
+    type Future = GetUser;
+    type Output = <GetUser as std::future::Future>::Output;
+    fn into_future(self) -> Self::Future {
+        Self::into_future(self)
     }
 }
diff --git a/sdk/data_cosmos/src/operations/list_collections.rs b/sdk/data_cosmos/src/operations/list_collections.rs
index d46fd1ab6f..0684bc1f89 100644
--- a/sdk/data_cosmos/src/operations/list_collections.rs
+++ b/sdk/data_cosmos/src/operations/list_collections.rs
@@ -2,40 +2,70 @@ use crate::headers::from_headers::*;
 use crate::prelude::*;
 use crate::resources::Collection;
 use crate::ResourceQuota;
-use azure_core::collect_pinned_stream;
 use azure_core::headers::{continuation_token_from_headers_optional, session_token_from_headers};
 use azure_core::prelude::*;
-use azure_core::Request as HttpRequest;
 use azure_core::Response as HttpResponse;
+use azure_core::{collect_pinned_stream, headers, Pageable};
 use chrono::{DateTime, Utc};
 
 #[derive(Debug, Clone)]
-pub struct ListCollectionsOptions {
+pub struct ListCollectionsBuilder {
+    client: DatabaseClient,
     consistency_level: Option<ConsistencyLevel>,
     max_item_count: MaxItemCount,
+    context: Context,
 }
 
-impl ListCollectionsOptions {
-    pub fn new() -> Self {
+impl ListCollectionsBuilder {
+    pub(crate) fn new(client: DatabaseClient) -> Self {
         Self {
+            client,
             max_item_count: MaxItemCount::new(-1),
             consistency_level: None,
+            context: Context::new(),
         }
     }
 
     setters! {
         consistency_level: ConsistencyLevel => Some(consistency_level),
         max_item_count: i32 => MaxItemCount::new(max_item_count),
+        context: Context => context,
     }
 
-    pub fn decorate_request(&self, request: &mut HttpRequest) -> crate::Result<()> {
-        azure_core::headers::add_optional_header2(&self.consistency_level, request)?;
-        azure_core::headers::add_mandatory_header2(&self.max_item_count, request)?;
+    pub fn into_stream(self) -> ListCollections {
+        let make_request = move |continuation: Option<String>| {
+            let this = self.clone();
+            let ctx = self.context.clone();
+            async move {
+                let mut request = this.client.cosmos_client().prepare_request_pipeline(
+                    &format!("dbs/{}/colls", this.client.database_name()),
+                    http::Method::GET,
+                );
 
-        Ok(())
+                azure_core::headers::add_optional_header2(&this.consistency_level, &mut request)?;
+                azure_core::headers::add_mandatory_header2(&this.max_item_count, &mut request)?;
+
+                if let Some(c) = continuation {
+                    let h = http::HeaderValue::from_str(c.as_str())
+                        .map_err(azure_core::HttpHeaderError::InvalidHeaderValue)?;
+                    request.headers_mut().append(headers::CONTINUATION, h);
+                }
+
+                let response = this
+                    .client
+                    .pipeline()
+                    .send(ctx.clone().insert(ResourceType::Collections), &mut request)
+                    .await?;
+                ListCollectionsResponse::try_from(response).await
+            }
+        };
+
+        Pageable::new(make_request)
     }
 }
 
+pub type ListCollections = Pageable<ListCollectionsResponse, crate::Error>;
+
 #[derive(Debug, Clone, PartialEq)]
 pub struct ListCollectionsResponse {
     pub rid: String,
@@ -90,3 +120,9 @@ impl ListCollectionsResponse {
         })
     }
 }
+
+impl Continuable for ListCollectionsResponse {
+    fn continuation(&self) -> Option<String> {
+        self.continuation_token.clone()
+    }
+}
diff --git a/sdk/data_cosmos/src/operations/list_databases.rs b/sdk/data_cosmos/src/operations/list_databases.rs
index cf7761bccc..c492a8addc 100644
--- a/sdk/data_cosmos/src/operations/list_databases.rs
+++ b/sdk/data_cosmos/src/operations/list_databases.rs
@@ -12,33 +12,33 @@ use azure_core::{collect_pinned_stream, prelude::*, Pageable, Response};
 use chrono::{DateTime, Utc};
 
 #[derive(Debug, Clone)]
-pub struct ListDatabases {
+pub struct ListDatabasesBuilder {
     client: CosmosClient,
     consistency_level: Option<ConsistencyLevel>,
     max_item_count: MaxItemCount,
-    context: Option<Context>,
+    context: Context,
 }
 
-impl ListDatabases {
-    pub fn new(client: CosmosClient) -> Self {
+impl ListDatabasesBuilder {
+    pub(crate) fn new(client: CosmosClient) -> Self {
         Self {
             client,
             consistency_level: None,
             max_item_count: MaxItemCount::new(-1),
-            context: None,
+            context: Context::new(),
         }
     }
 
     setters! {
         consistency_level: ConsistencyLevel => Some(consistency_level),
         max_item_count: i32 => MaxItemCount::new(max_item_count),
-        context: Context => Some(context),
+        context: Context => context,
     }
 
-    pub fn into_stream(self) -> Pageable<ListDatabasesResponse, azure_core::error::Error> {
+    pub fn into_stream(self) -> ListDatabases {
         let make_request = move |continuation: Option<String>| {
             let this = self.clone();
-            let ctx = self.context.clone().unwrap_or_default();
+            let ctx = self.context.clone();
             async move {
                 let mut request = this
                     .client
@@ -81,6 +81,8 @@ impl ListDatabases {
     }
 }
 
+pub type ListDatabases = Pageable<ListDatabasesResponse, azure_core::error::Error>;
+
 #[derive(Clone, PartialEq, PartialOrd, Debug)]
 pub struct ListDatabasesResponse {
     pub rid: String,
diff --git a/sdk/data_cosmos/src/operations/list_users.rs b/sdk/data_cosmos/src/operations/list_users.rs
index e7f0e5eaa5..01e6d52e59 100644
--- a/sdk/data_cosmos/src/operations/list_users.rs
+++ b/sdk/data_cosmos/src/operations/list_users.rs
@@ -5,36 +5,68 @@ use azure_core::{
     collect_pinned_stream,
     headers::{continuation_token_from_headers_optional, session_token_from_headers},
     prelude::MaxItemCount,
-    Request as HttpRequest, Response as HttpResponse, SessionToken,
+    Response as HttpResponse, SessionToken,
 };
+use azure_core::{headers, Context, Continuable, Pageable};
 
 #[derive(Debug, Clone)]
-pub struct ListUsersOptions {
+pub struct ListUsersBuilder {
+    client: DatabaseClient,
     consistency_level: Option<ConsistencyLevel>,
     max_item_count: MaxItemCount,
+    context: Context,
 }
 
-impl ListUsersOptions {
-    pub fn new() -> Self {
+impl ListUsersBuilder {
+    pub(crate) fn new(client: DatabaseClient) -> Self {
         Self {
+            client,
             consistency_level: None,
             max_item_count: MaxItemCount::new(-1),
+            context: Context::new(),
         }
     }
 
     setters! {
         consistency_level: ConsistencyLevel => Some(consistency_level),
         max_item_count: i32 => MaxItemCount::new(max_item_count),
+        context: Context => context,
     }
 
-    pub(crate) fn decorate_request(&self, request: &mut HttpRequest) -> crate::Result<()> {
-        azure_core::headers::add_optional_header2(&self.consistency_level, request)?;
-        azure_core::headers::add_mandatory_header2(&self.max_item_count, request)?;
+    pub fn into_stream(self) -> ListUsers {
+        let make_request = move |continuation: Option<String>| {
+            let this = self.clone();
+            let ctx = self.context.clone();
+            async move {
+                let mut request = this.client.cosmos_client().prepare_request_pipeline(
+                    &format!("dbs/{}/users", this.client.database_name()),
+                    http::Method::GET,
+                );
 
-        Ok(())
+                azure_core::headers::add_optional_header2(&this.consistency_level, &mut request)?;
+                azure_core::headers::add_mandatory_header2(&this.max_item_count, &mut request)?;
+
+                if let Some(c) = continuation {
+                    let h = http::HeaderValue::from_str(c.as_str())
+                        .map_err(azure_core::HttpHeaderError::InvalidHeaderValue)?;
+                    request.headers_mut().append(headers::CONTINUATION, h);
+                }
+
+                let response = this
+                    .client
+                    .pipeline()
+                    .send(ctx.clone().insert(ResourceType::Users), &mut request)
+                    .await?;
+                ListUsersResponse::try_from(response).await
+            }
+        };
+
+        Pageable::new(make_request)
     }
 }
 
+pub type ListUsers = Pageable<ListUsersResponse, crate::Error>;
+
 #[derive(Debug, Clone, PartialEq)]
 pub struct ListUsersResponse {
     pub users: Vec<User>,
@@ -84,3 +116,9 @@ impl IntoIterator for ListUsersResponse {
         self.users.into_iter()
     }
 }
+
+impl Continuable for ListUsersResponse {
+    fn continuation(&self) -> Option<String> {
+        self.continuation_token.clone()
+    }
+}
diff --git a/sdk/data_cosmos/src/operations/replace_collection.rs b/sdk/data_cosmos/src/operations/replace_collection.rs
index a38afc170f..87490939d1 100644
--- a/sdk/data_cosmos/src/operations/replace_collection.rs
+++ b/sdk/data_cosmos/src/operations/replace_collection.rs
@@ -4,45 +4,75 @@ use crate::resources::collection::{IndexingPolicy, PartitionKey};
 use azure_core::headers::{
     content_type_from_headers, etag_from_headers, session_token_from_headers,
 };
-use azure_core::{collect_pinned_stream, Request as HttpRequest, Response as HttpResponse};
+use azure_core::{collect_pinned_stream, Context, Response as HttpResponse};
 use chrono::{DateTime, Utc};
 
 #[derive(Debug, Clone)]
-pub struct ReplaceCollectionOptions {
+pub struct ReplaceCollectionBuilder {
+    client: CollectionClient,
     partition_key: PartitionKey,
     consistency_level: Option<ConsistencyLevel>,
     indexing_policy: Option<IndexingPolicy>,
+    context: Context,
 }
 
-impl ReplaceCollectionOptions {
-    pub fn new<P: Into<PartitionKey>>(partition_key: P) -> Self {
+impl ReplaceCollectionBuilder {
+    pub(crate) fn new(client: CollectionClient, partition_key: PartitionKey) -> Self {
         Self {
-            partition_key: partition_key.into(),
+            client,
+            partition_key,
             consistency_level: None,
             indexing_policy: None,
+            context: Context::new(),
         }
     }
 
     setters! {
         consistency_level: ConsistencyLevel => Some(consistency_level),
         indexing_policy: IndexingPolicy => Some(indexing_policy),
+        context: Context => context,
     }
 
-    pub(crate) fn decorate_request(
-        &self,
-        request: &mut HttpRequest,
-        collection_name: &str,
-    ) -> crate::Result<()> {
-        azure_core::headers::add_optional_header2(&self.consistency_level, request)?;
+    pub fn into_future(self) -> ReplaceCollection {
+        Box::pin(async move {
+            let mut request = self
+                .client
+                .prepare_request_with_collection_name(http::Method::PUT);
 
-        let collection = ReplaceCollectionBody {
-            id: collection_name,
-            indexing_policy: &self.indexing_policy,
-            partition_key: &self.partition_key,
-        };
+            azure_core::headers::add_optional_header2(&self.consistency_level, &mut request)?;
 
-        request.set_body(bytes::Bytes::from(serde_json::to_string(&collection)?).into());
-        Ok(())
+            let collection = ReplaceCollectionBody {
+                id: self.client.collection_name(),
+                indexing_policy: &self.indexing_policy,
+                partition_key: &self.partition_key,
+            };
+
+            request.set_body(bytes::Bytes::from(serde_json::to_string(&collection)?).into());
+
+            let response = self
+                .client
+                .pipeline()
+                .send(
+                    self.context.clone().insert(ResourceType::Collections),
+                    &mut request,
+                )
+                .await?;
+
+            ReplaceCollectionResponse::try_from(response).await
+        })
+    }
+}
+
+/// The future returned by calling `into_future` on the builder.
+pub type ReplaceCollection =
+    futures::future::BoxFuture<'static, crate::Result<ReplaceCollectionResponse>>;
+
+#[cfg(feature = "into_future")]
+impl std::future::IntoFuture for ReplaceCollectionBuilder {
+    type Future = ReplaceCollection;
+    type Output = <ReplaceCollection as std::future::Future>::Output;
+    fn into_future(self) -> Self::Future {
+        Self::into_future(self)
     }
 }
 
diff --git a/sdk/data_cosmos/src/operations/replace_document.rs b/sdk/data_cosmos/src/operations/replace_document.rs
index 4c5a1a7689..7882b0b453 100644
--- a/sdk/data_cosmos/src/operations/replace_document.rs
+++ b/sdk/data_cosmos/src/operations/replace_document.rs
@@ -7,40 +7,45 @@ use crate::ResourceQuota;
 use azure_core::headers::session_token_from_headers;
 use azure_core::prelude::*;
 use azure_core::SessionToken;
-use azure_core::{collect_pinned_stream, Request as HttpRequest, Response as HttpResponse};
+use azure_core::{collect_pinned_stream, Response as HttpResponse};
 use chrono::{DateTime, Utc};
 use serde::Serialize;
 
 #[derive(Debug, Clone)]
-pub struct ReplaceDocumentOptions {
+pub struct ReplaceDocumentBuilder<D> {
+    client: DocumentClient,
+    document: D,
     partition_key: Option<String>,
     indexing_directive: IndexingDirective,
     if_match_condition: Option<IfMatchCondition>,
     if_modified_since: Option<IfModifiedSince>,
     consistency_level: Option<ConsistencyLevel>,
     allow_tentative_writes: TentativeWritesAllowance,
+    context: Context,
 }
 
-impl ReplaceDocumentOptions {
-    pub fn new() -> Self {
+impl<D: Serialize + Send + 'static> ReplaceDocumentBuilder<D> {
+    pub(crate) fn new(client: DocumentClient, document: D) -> Self {
         Self {
+            client,
+            document,
             partition_key: None,
             indexing_directive: IndexingDirective::Default,
             if_match_condition: None,
             if_modified_since: None,
             consistency_level: None,
             allow_tentative_writes: TentativeWritesAllowance::Deny,
+            context: Context::new(),
         }
     }
-}
 
-impl ReplaceDocumentOptions {
     setters! {
         consistency_level: ConsistencyLevel => Some(consistency_level),
         if_match_condition: IfMatchCondition => Some(if_match_condition),
         if_modified_since: DateTime<Utc> => Some(IfModifiedSince::new(if_modified_since)),
         allow_tentative_writes: TentativeWritesAllowance,
         indexing_directive: IndexingDirective,
+        context: Context => context,
     }
 
     pub fn partition_key<T: Serialize>(&mut self, partition_key: &T) -> crate::Result<()> {
@@ -48,31 +53,52 @@ impl ReplaceDocumentOptions {
         Ok(())
     }
 
-    pub fn decorate_request<'b, D>(
-        &self,
-        request: &mut HttpRequest,
-        document: &'b D,
-        serialized_partition_key: &str,
-    ) -> crate::Result<()>
-    where
-        D: Serialize,
-    {
-        let partition_key = self
-            .partition_key
-            .as_deref()
-            .unwrap_or(serialized_partition_key);
-        add_as_partition_key_header_serialized2(partition_key, request);
-
-        azure_core::headers::add_mandatory_header2(&self.indexing_directive, request)?;
-        azure_core::headers::add_optional_header2(&self.if_match_condition, request)?;
-        azure_core::headers::add_optional_header2(&self.if_modified_since, request)?;
-        azure_core::headers::add_optional_header2(&self.consistency_level, request)?;
-        azure_core::headers::add_mandatory_header2(&self.allow_tentative_writes, request)?;
-
-        let serialized = azure_core::to_json(document)?;
-        request.set_body(serialized.into());
+    pub fn into_future(self) -> ReplaceDocument {
+        Box::pin(async move {
+            let mut request = self
+                .client
+                .prepare_request_pipeline_with_document_name(http::Method::PUT);
 
-        Ok(())
+            let partition_key = self
+                .partition_key
+                .as_deref()
+                .unwrap_or_else(|| self.client.partition_key_serialized());
+            add_as_partition_key_header_serialized2(partition_key, &mut request);
+
+            azure_core::headers::add_mandatory_header2(&self.indexing_directive, &mut request)?;
+            azure_core::headers::add_optional_header2(&self.if_match_condition, &mut request)?;
+            azure_core::headers::add_optional_header2(&self.if_modified_since, &mut request)?;
+            azure_core::headers::add_optional_header2(&self.consistency_level, &mut request)?;
+            azure_core::headers::add_mandatory_header2(&self.allow_tentative_writes, &mut request)?;
+
+            let serialized = azure_core::to_json(&self.document)?;
+            request.set_body(serialized.into());
+
+            let response = self
+                .client
+                .cosmos_client()
+                .pipeline()
+                .send(
+                    self.context.clone().insert(ResourceType::Documents),
+                    &mut request,
+                )
+                .await?;
+
+            ReplaceDocumentResponse::try_from(response).await
+        })
+    }
+}
+
+/// The future returned by calling `into_future` on the builder.
+pub type ReplaceDocument =
+    futures::future::BoxFuture<'static, crate::Result<ReplaceDocumentResponse>>;
+
+#[cfg(feature = "into_future")]
+impl std::future::IntoFuture for ReplaceDocumentBuilder {
+    type Future = ReplaceDocument;
+    type Output = <ReplaceDocument as std::future::Future>::Output;
+    fn into_future(self) -> Self::Future {
+        Self::into_future(self)
     }
 }
 
diff --git a/sdk/data_cosmos/tests/collection_operations.rs b/sdk/data_cosmos/tests/collection_operations.rs
index aeb1727338..d9c6e2f576 100644
--- a/sdk/data_cosmos/tests/collection_operations.rs
+++ b/sdk/data_cosmos/tests/collection_operations.rs
@@ -1,6 +1,5 @@
 #![cfg(feature = "mock_transport_framework")]
 
-use azure_core::prelude::*;
 use azure_data_cosmos::prelude::*;
 use azure_data_cosmos::resources::collection::*;
 use std::error::Error;
@@ -15,7 +14,6 @@ async fn collection_operations() -> Result<(), BoxedError> {
 
     let client = setup::initialize("collection_operations")?;
     let database_name = "test-collection-operations";
-    let context = Context::new();
 
     client.create_database(database_name).into_future().await?;
 
@@ -75,10 +73,9 @@ async fn collection_operations() -> Result<(), BoxedError> {
 
     // replace collection!
     let replace_collection_response = collection_client
-        .replace_collection(
-            context.clone(),
-            ReplaceCollectionOptions::new("/id").indexing_policy(new_indexing_policy),
-        )
+        .replace_collection("/id")
+        .indexing_policy(new_indexing_policy)
+        .into_future()
         .await?;
 
     assert_eq!(replace_collection_response.collection.id, collection_name);
diff --git a/sdk/data_cosmos/tests/cosmos_collection.rs b/sdk/data_cosmos/tests/cosmos_collection.rs
index e1e47595df..52d523f81e 100644
--- a/sdk/data_cosmos/tests/cosmos_collection.rs
+++ b/sdk/data_cosmos/tests/cosmos_collection.rs
@@ -1,7 +1,6 @@
 #![cfg(all(test, feature = "test_e2e"))]
 mod setup;
 
-use azure_core::prelude::*;
 use azure_data_cosmos::prelude::*;
 use azure_data_cosmos::resources::collection::*;
 use futures::stream::StreamExt;
@@ -27,12 +26,11 @@ async fn create_and_delete_collection() {
         .into_future()
         .await
         .unwrap();
-    let collections =
-        Box::pin(database_client.list_collections(Context::new(), ListCollectionsOptions::new()))
-            .next()
-            .await
-            .unwrap()
-            .unwrap();
+    let collections = Box::pin(database_client.list_collections().into_stream())
+        .next()
+        .await
+        .unwrap()
+        .unwrap();
     assert!(collections.collections.len() == 1);
 
     // try to get the previously created collection
@@ -60,12 +58,11 @@ async fn create_and_delete_collection() {
         .into_future()
         .await
         .unwrap();
-    let collections =
-        Box::pin(database_client.list_collections(Context::new(), ListCollectionsOptions::new()))
-            .next()
-            .await
-            .unwrap()
-            .unwrap();
+    let collections = Box::pin(database_client.list_collections().into_stream())
+        .next()
+        .await
+        .unwrap()
+        .unwrap();
     assert!(collections.collections.len() == 0);
 
     database_client
@@ -105,12 +102,11 @@ async fn replace_collection() {
         .await
         .unwrap();
 
-    let collections =
-        Box::pin(database_client.list_collections(Context::new(), ListCollectionsOptions::new()))
-            .next()
-            .await
-            .unwrap()
-            .unwrap();
+    let collections = Box::pin(database_client.list_collections().into_stream())
+        .next()
+        .await
+        .unwrap()
+        .unwrap();
     assert_eq!(collections.collections.len(), 1);
     assert_eq!(
         collection.collection.indexing_policy,
@@ -145,19 +141,17 @@ async fn replace_collection() {
         .into_collection_client(COLLECTION_NAME);
 
     let _replace_collection_response = collection_client
-        .replace_collection(
-            Context::new(),
-            ReplaceCollectionOptions::new("/id").indexing_policy(new_ip),
-        )
+        .replace_collection("/id")
+        .indexing_policy(new_ip)
+        .into_future()
         .await
         .unwrap();
 
-    let collections =
-        Box::pin(database_client.list_collections(Context::new(), ListCollectionsOptions::new()))
-            .next()
-            .await
-            .unwrap()
-            .unwrap();
+    let collections = Box::pin(database_client.list_collections().into_stream())
+        .next()
+        .await
+        .unwrap()
+        .unwrap();
     assert_eq!(collections.collections.len(), 1);
     let eps: Vec<&ExcludedPath> = collections.collections[0]
         .indexing_policy
diff --git a/sdk/data_cosmos/tests/cosmos_document.rs b/sdk/data_cosmos/tests/cosmos_document.rs
index 7154dc9b60..c970225ed7 100644
--- a/sdk/data_cosmos/tests/cosmos_document.rs
+++ b/sdk/data_cosmos/tests/cosmos_document.rs
@@ -1,6 +1,4 @@
 #![cfg(all(test, feature = "test_e2e"))]
-use azure_core::Context;
-use azure_data_cosmos::prelude::GetDocumentOptions;
 use serde::{Deserialize, Serialize};
 
 mod setup;
@@ -85,7 +83,8 @@ async fn create_and_delete_document() {
         .unwrap();
 
     let document_after_get = document_client
-        .get_document::<MyDocument>(Context::new(), GetDocumentOptions::new())
+        .get_document()
+        .into_future::<MyDocument>()
         .await
         .unwrap();
 
@@ -252,18 +251,15 @@ async fn replace_document() {
         .clone()
         .into_document_client(document_data.id.clone(), &document_data.id)
         .unwrap()
-        .replace_document(
-            Context::new(),
-            &document_data,
-            ReplaceDocumentOptions::new()
-                .consistency_level(ConsistencyLevel::from(&documents))
-                .if_match_condition(IfMatchCondition::Match(
-                    documents.documents[0]
-                        .document_attributes
-                        .etag()
-                        .to_string(),
-                )),
-        )
+        .replace_document(document_data)
+        .consistency_level(ConsistencyLevel::from(&documents))
+        .if_match_condition(IfMatchCondition::Match(
+            documents.documents[0]
+                .document_attributes
+                .etag()
+                .to_string(),
+        ))
+        .into_future()
         .await
         .unwrap();
 
@@ -272,7 +268,8 @@ async fn replace_document() {
         .into_document_client(DOCUMENT_NAME, &DOCUMENT_NAME)
         .unwrap();
     let document_after_get = document_client
-        .get_document::<MyDocument>(Context::new(), GetDocumentOptions::new())
+        .get_document()
+        .into_future::<MyDocument>()
         .await
         .unwrap();
 
diff --git a/sdk/data_cosmos/tests/create_database_and_collection.rs b/sdk/data_cosmos/tests/create_database_and_collection.rs
index b84948754e..680141e6ca 100644
--- a/sdk/data_cosmos/tests/create_database_and_collection.rs
+++ b/sdk/data_cosmos/tests/create_database_and_collection.rs
@@ -1,6 +1,5 @@
 #![cfg(feature = "mock_transport_framework")]
 
-use azure_core::Context;
 use azure_data_cosmos::prelude::*;
 use futures::stream::StreamExt;
 use std::error::Error;
@@ -15,7 +14,6 @@ async fn create_database_and_collection() -> Result<(), BoxedError> {
 
     let client = setup::initialize("create_database_and_collection")?;
     let database_name = "test-create-database-and-collection";
-    let context = Context::new();
 
     // create database!
     log::info!("Creating a database with name '{}'...", database_name);
@@ -39,11 +37,10 @@ async fn create_database_and_collection() -> Result<(), BoxedError> {
 
     // list collections!
     log::info!("Listing all collections...");
-    let collections =
-        Box::pin(db_client.list_collections(context.clone(), ListCollectionsOptions::new()))
-            .next()
-            .await
-            .expect("No collection page")?;
+    let collections = Box::pin(db_client.list_collections().into_stream())
+        .next()
+        .await
+        .expect("No collection page")?;
     assert_eq!(collections.count, 1);
     log::info!("Successfully listed collections");
     log::debug!("The list_collection response: {:#?}", collections);
diff --git a/sdk/data_cosmos/tests/user.rs b/sdk/data_cosmos/tests/user.rs
index 08519a10ad..753ade7e58 100644
--- a/sdk/data_cosmos/tests/user.rs
+++ b/sdk/data_cosmos/tests/user.rs
@@ -34,18 +34,15 @@ async fn users() {
 
     let _create_user_response = user_client.create_user().into_future().await.unwrap();
 
-    let list_users_response =
-        Box::pin(database_client.list_users(Context::new(), ListUsersOptions::new()))
-            .next()
-            .await
-            .unwrap()
-            .unwrap();
+    let list_users_response = Box::pin(database_client.list_users().into_stream())
+        .next()
+        .await
+        .unwrap()
+        .unwrap();
 
     assert_eq!(list_users_response.users.len(), 1);
 
-    let get_user_response = user_client
-        .get_user(Context::new(), GetUserOptions::new())
-        .await;
+    let get_user_response = user_client.get_user().into_future().await;
     assert!(get_user_response.is_ok());
     let retrieved_user = get_user_response.unwrap();
     assert_eq!(retrieved_user.user.id, USER_NAME);
@@ -59,24 +56,22 @@ async fn users() {
         .await
         .unwrap();
 
-    let list_users_response =
-        Box::pin(database_client.list_users(Context::new(), ListUsersOptions::new()))
-            .next()
-            .await
-            .unwrap()
-            .unwrap();
+    let list_users_response = Box::pin(database_client.list_users().into_stream())
+        .next()
+        .await
+        .unwrap()
+        .unwrap();
     assert_eq!(list_users_response.users.len(), 1);
 
     let user_client = database_client.clone().into_user_client(USER_NAME_REPLACED);
 
     let _delete_user_response = user_client.delete_user().into_future().await.unwrap();
 
-    let list_users_response =
-        Box::pin(database_client.list_users(Context::new(), ListUsersOptions::new()))
-            .next()
-            .await
-            .unwrap()
-            .unwrap();
+    let list_users_response = Box::pin(database_client.list_users().into_stream())
+        .next()
+        .await
+        .unwrap()
+        .unwrap();
     assert_eq!(list_users_response.users.len(), 0);
 
     // delete the database