Add PaginatedListStore (#371)

* Add PaginatedListStore

* Format

* More docs

* Format

* Fix empty Azure pagination token

Author: tustvold

src/aws/client.rs

@@ -33,11 +33,11 @@ use crate::client::s3::{
     InitiateMultipartUploadResult, ListResponse, PartMetadata,
 };
 use crate::client::{GetOptionsExt, HttpClient, HttpError, HttpResponse};
+use crate::list::{PaginatedListOptions, PaginatedListResult};
 use crate::multipart::PartId;
-use crate::path::DELIMITER;
 use crate::{
-    Attribute, Attributes, ClientOptions, GetOptions, ListResult, MultipartId, Path,
-    PutMultipartOpts, PutPayload, PutResult, Result, RetryConfig, TagSet,
+    Attribute, Attributes, ClientOptions, GetOptions, MultipartId, Path, PutMultipartOpts,
+    PutPayload, PutResult, Result, RetryConfig, TagSet,
 };
 use async_trait::async_trait;
 use base64::prelude::BASE64_STANDARD;
@@ -877,21 +877,19 @@ impl ListClient for Arc<S3Client> {
     async fn list_request(
         &self,
         prefix: Option<&str>,
-        delimiter: bool,
-        token: Option<&str>,
-        offset: Option<&str>,
-    ) -> Result<(ListResult, Option<String>)> {
+        opts: PaginatedListOptions,
+    ) -> Result<PaginatedListResult> {
         let credential = self.config.get_session_credential().await?;
         let url = self.config.bucket_endpoint.clone();
 
         let mut query = Vec::with_capacity(4);
 
-        if let Some(token) = token {
-            query.push(("continuation-token", token))
+        if let Some(token) = &opts.page_token {
+            query.push(("continuation-token", token.as_ref()))
         }
 
-        if delimiter {
-            query.push(("delimiter", DELIMITER))
+        if let Some(d) = &opts.delimiter {
+            query.push(("delimiter", d.as_ref()))
         }
 
         query.push(("list-type", "2"));
@@ -900,13 +898,20 @@ impl ListClient for Arc<S3Client> {
             query.push(("prefix", prefix))
         }
 
-        if let Some(offset) = offset {
-            query.push(("start-after", offset))
+        if let Some(offset) = &opts.offset {
+            query.push(("start-after", offset.as_ref()))
+        }
+
+        let max_keys_str;
+        if let Some(max_keys) = &opts.max_keys {
+            max_keys_str = max_keys.to_string();
+            query.push(("max-keys", max_keys_str.as_ref()))
         }
 
         let response = self
             .client
             .request(Method::GET, &url)
+            .extensions(opts.extensions)
             .query(&query)
             .with_aws_sigv4(credential.authorizer(), None)
             .send_retry(&self.config.retry_config)
@@ -922,7 +927,10 @@ impl ListClient for Arc<S3Client> {
 
         let token = response.next_continuation_token.take();
 
-        Ok((response.try_into()?, token))
+        Ok(PaginatedListResult {
+            result: response.try_into()?,
+            page_token: token,
+        })
     }
 }
 

src/aws/mod.rs

@@ -38,7 +38,7 @@ use url::Url;
 
 use crate::aws::client::{CompleteMultipartMode, PutPartPayload, RequestError, S3Client};
 use crate::client::get::GetClientExt;
-use crate::client::list::ListClientExt;
+use crate::client::list::{ListClient, ListClientExt};
 use crate::client::CredentialProvider;
 use crate::multipart::{MultipartStore, PartId};
 use crate::signer::Signer;
@@ -78,6 +78,7 @@ const STORE: &str = "S3";
 /// [`CredentialProvider`] for [`AmazonS3`]
 pub type AwsCredentialProvider = Arc<dyn CredentialProvider<Credential = AwsCredential>>;
 use crate::client::parts::Parts;
+use crate::list::{PaginatedListOptions, PaginatedListResult, PaginatedListStore};
 pub use credential::{AwsAuthorizer, AwsCredential};
 
 /// Interface for [Amazon S3](https://aws.amazon.com/s3/).
@@ -496,6 +497,17 @@ impl MultipartStore for AmazonS3 {
     }
 }
 
+#[async_trait]
+impl PaginatedListStore for AmazonS3 {
+    async fn list_paginated(
+        &self,
+        prefix: Option<&str>,
+        opts: PaginatedListOptions,
+    ) -> Result<PaginatedListResult> {
+        self.client.list_request(prefix, opts).await
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -594,6 +606,7 @@ mod tests {
         signing(&integration).await;
         s3_encryption(&integration).await;
         put_get_attributes(&integration).await;
+        list_paginated(&integration, &integration).await;
 
         // Object tagging is not supported by S3 Express One Zone
         if config.session_provider.is_none() {

src/azure/client.rs

@@ -24,8 +24,8 @@ use crate::client::header::{get_put_result, HeaderConfig};
 use crate::client::list::ListClient;
 use crate::client::retry::RetryExt;
 use crate::client::{GetOptionsExt, HttpClient, HttpError, HttpRequest, HttpResponse};
+use crate::list::{PaginatedListOptions, PaginatedListResult};
 use crate::multipart::PartId;
-use crate::path::DELIMITER;
 use crate::util::{deserialize_rfc1123, GetRange};
 use crate::{
     Attribute, Attributes, ClientOptions, GetOptions, ListResult, ObjectMeta, Path, PutMode,
@@ -961,11 +961,13 @@ impl ListClient for Arc<AzureClient> {
     async fn list_request(
         &self,
         prefix: Option<&str>,
-        delimiter: bool,
-        token: Option<&str>,
-        offset: Option<&str>,
-    ) -> Result<(ListResult, Option<String>)> {
-        assert!(offset.is_none()); // Not yet supported
+        opts: PaginatedListOptions,
+    ) -> Result<PaginatedListResult> {
+        if opts.offset.is_some() {
+            return Err(crate::Error::NotSupported {
+                source: "Azure does not support listing with offsets".into(),
+            });
+        }
 
         let credential = self.get_credential().await?;
         let url = self.config.path_url(&Path::default());
@@ -978,21 +980,29 @@ impl ListClient for Arc<AzureClient> {
             query.push(("prefix", prefix))
         }
 
-        if delimiter {
-            query.push(("delimiter", DELIMITER))
+        if let Some(delimiter) = &opts.delimiter {
+            query.push(("delimiter", delimiter.as_ref()))
         }
 
-        if let Some(token) = token {
-            query.push(("marker", token))
+        if let Some(token) = &opts.page_token {
+            query.push(("marker", token.as_ref()))
+        }
+
+        let max_keys_str;
+        if let Some(max_keys) = &opts.max_keys {
+            max_keys_str = max_keys.to_string();
+            query.push(("maxresults", max_keys_str.as_ref()))
         }
 
         let sensitive = credential
             .as_deref()
             .map(|c| c.sensitive_request())
             .unwrap_or_default();
+
         let response = self
             .client
             .get(url.as_str())
+            .extensions(opts.extensions)
             .query(&query)
             .with_azure_authorization(&credential, &self.config.account)
             .retryable(&self.config.retry_config)
@@ -1008,9 +1018,12 @@ impl ListClient for Arc<AzureClient> {
         let mut response: ListResultInternal = quick_xml::de::from_reader(response.reader())
             .map_err(|source| Error::InvalidListResponse { source })?;
 
-        let token = response.next_marker.take();
+        let token = response.next_marker.take().filter(|x| !x.is_empty());
 
-        Ok((to_list_result(response, prefix)?, token))
+        Ok(PaginatedListResult {
+            result: to_list_result(response, prefix)?,
+            page_token: token,
+        })
     }
 }
 

src/azure/mod.rs

@@ -38,7 +38,7 @@ use std::time::Duration;
 use url::Url;
 
 use crate::client::get::GetClientExt;
-use crate::client::list::ListClientExt;
+use crate::client::list::{ListClient, ListClientExt};
 use crate::client::CredentialProvider;
 pub use credential::{authority_hosts, AzureAccessKey, AzureAuthorizer};
 
@@ -50,6 +50,7 @@ mod credential;
 pub type AzureCredentialProvider = Arc<dyn CredentialProvider<Credential = AzureCredential>>;
 use crate::azure::client::AzureClient;
 use crate::client::parts::Parts;
+use crate::list::{PaginatedListOptions, PaginatedListResult, PaginatedListStore};
 pub use builder::{AzureConfigKey, MicrosoftAzureBuilder};
 pub use credential::AzureCredential;
 
@@ -292,6 +293,17 @@ impl MultipartStore for MicrosoftAzure {
     }
 }
 
+#[async_trait]
+impl PaginatedListStore for MicrosoftAzure {
+    async fn list_paginated(
+        &self,
+        prefix: Option<&str>,
+        opts: PaginatedListOptions,
+    ) -> Result<PaginatedListResult> {
+        self.client.list_request(prefix, opts).await
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -316,6 +328,7 @@ mod tests {
         multipart_race_condition(&integration, false).await;
         multipart_out_of_order(&integration).await;
         signing(&integration).await;
+        list_paginated(&integration, &integration).await;
 
         let validate = !integration.client.config().disable_tagging;
         tagging(

src/client/list.rs

@@ -16,12 +16,14 @@
 // under the License.
 
 use crate::client::pagination::stream_paginated;
-use crate::path::Path;
+use crate::list::{PaginatedListOptions, PaginatedListResult};
+use crate::path::{Path, DELIMITER};
 use crate::Result;
 use crate::{ListResult, ObjectMeta};
 use async_trait::async_trait;
 use futures::stream::BoxStream;
 use futures::{StreamExt, TryStreamExt};
+use std::borrow::Cow;
 use std::collections::BTreeSet;
 
 /// A client that can perform paginated list requests
@@ -30,10 +32,8 @@ pub(crate) trait ListClient: Send + Sync + 'static {
     async fn list_request(
         &self,
         prefix: Option<&str>,
-        delimiter: bool,
-        token: Option<&str>,
-        offset: Option<&str>,
-    ) -> Result<(ListResult, Option<String>)>;
+        options: PaginatedListOptions,
+    ) -> Result<PaginatedListResult>;
 }
 
 /// Extension trait for [`ListClient`] that adds common listing functionality
@@ -69,21 +69,23 @@ impl<T: ListClient + Clone> ListClientExt for T {
         let offset = offset.map(|x| x.to_string());
         let prefix = prefix
             .filter(|x| !x.as_ref().is_empty())
-            .map(|p| format!("{}{}", p.as_ref(), crate::path::DELIMITER));
-
+            .map(|p| format!("{}{}", p.as_ref(), DELIMITER));
         stream_paginated(
             self.clone(),
             (prefix, offset),
-            move |client, (prefix, offset), token| async move {
-                let (r, next_token) = client
+            move |client, (prefix, offset), page_token| async move {
+                let r = client
                     .list_request(
                         prefix.as_deref(),
-                        delimiter,
-                        token.as_deref(),
-                        offset.as_deref(),
+                        PaginatedListOptions {
+                            offset: offset.clone(),
+                            delimiter: delimiter.then_some(Cow::Borrowed(DELIMITER)),
+                            page_token,
+                            ..Default::default()
+                        },
                     )
                     .await?;
-                Ok((r, (prefix, offset), next_token))
+                Ok((r.result, (prefix, offset), r.page_token))
             },
         )
         .boxed()

src/gcp/client.rs

@@ -27,12 +27,13 @@ use crate::client::s3::{
 use crate::client::{GetOptionsExt, HttpClient, HttpError, HttpResponse};
 use crate::gcp::credential::CredentialExt;
 use crate::gcp::{GcpCredential, GcpCredentialProvider, GcpSigningCredentialProvider, STORE};
+use crate::list::{PaginatedListOptions, PaginatedListResult};
 use crate::multipart::PartId;
-use crate::path::{Path, DELIMITER};
+use crate::path::Path;
 use crate::util::hex_encode;
 use crate::{
-    Attribute, Attributes, ClientOptions, GetOptions, ListResult, MultipartId, PutMode,
-    PutMultipartOpts, PutOptions, PutPayload, PutResult, Result, RetryConfig,
+    Attribute, Attributes, ClientOptions, GetOptions, MultipartId, PutMode, PutMultipartOpts,
+    PutOptions, PutPayload, PutResult, Result, RetryConfig,
 };
 use async_trait::async_trait;
 use base64::prelude::BASE64_STANDARD;
@@ -652,38 +653,43 @@ impl ListClient for Arc<GoogleCloudStorageClient> {
     async fn list_request(
         &self,
         prefix: Option<&str>,
-        delimiter: bool,
-        page_token: Option<&str>,
-        offset: Option<&str>,
-    ) -> Result<(ListResult, Option<String>)> {
+        opts: PaginatedListOptions,
+    ) -> Result<PaginatedListResult> {
         let credential = self.get_credential().await?;
         let url = format!("{}/{}", self.config.base_url, self.bucket_name_encoded);
 
         let mut query = Vec::with_capacity(5);
         query.push(("list-type", "2"));
-        if delimiter {
-            query.push(("delimiter", DELIMITER))
+        if let Some(delimiter) = &opts.delimiter {
+            query.push(("delimiter", delimiter.as_ref()))
         }
 
-        if let Some(prefix) = &prefix {
+        if let Some(prefix) = prefix {
             query.push(("prefix", prefix))
         }
 
-        if let Some(page_token) = page_token {
+        if let Some(page_token) = &opts.page_token {
             query.push(("continuation-token", page_token))
         }
 
         if let Some(max_results) = &self.max_list_results {
             query.push(("max-keys", max_results))
         }
 
-        if let Some(offset) = offset {
-            query.push(("start-after", offset))
+        if let Some(offset) = &opts.offset {
+            query.push(("start-after", offset.as_ref()))
+        }
+
+        let max_keys_str;
+        if let Some(max_keys) = &opts.max_keys {
+            max_keys_str = max_keys.to_string();
+            query.push(("max-keys", max_keys_str.as_ref()))
         }
 
         let response = self
             .client
             .request(Method::GET, url)
+            .extensions(opts.extensions)
             .query(&query)
             .with_bearer_auth(credential.as_deref())
             .send_retry(&self.config.retry_config)
@@ -698,6 +704,9 @@ impl ListClient for Arc<GoogleCloudStorageClient> {
             .map_err(|source| Error::InvalidListResponse { source })?;
 
         let token = response.next_continuation_token.take();
-        Ok((response.try_into()?, token))
+        Ok(PaginatedListResult {
+            result: response.try_into()?,
+            page_token: token,
+        })
     }
 }