refactor: introduce ObjectStoreExt trait

src/aws/mod.rs

@@ -499,6 +499,7 @@ mod tests {
     use crate::integration::*;
     use crate::tests::*;
     use crate::ClientOptions;
+    use crate::ObjectStoreExt;
     use base64::prelude::BASE64_STANDARD;
     use base64::Engine;
     use http::HeaderMap;

src/azure/mod.rs

@@ -309,6 +309,7 @@ mod tests {
     use super::*;
     use crate::integration::*;
     use crate::tests::*;
+    use crate::ObjectStoreExt;
     use bytes::Bytes;
 
     #[tokio::test]

src/buffered.rs

@@ -210,12 +210,12 @@ impl AsyncBufRead for BufReader {
 
 /// An async buffered writer compatible with the tokio IO traits
 ///
-/// This writer adaptively uses [`ObjectStore::put`] or
+/// This writer adaptively uses [`ObjectStore::put_opts`] or
 /// [`ObjectStore::put_multipart`] depending on the amount of data that has
 /// been written.
 ///
 /// Up to `capacity` bytes will be buffered in memory, and flushed on shutdown
-/// using [`ObjectStore::put`]. If `capacity` is exceeded, data will instead be
+/// using [`ObjectStore::put_opts`]. If `capacity` is exceeded, data will instead be
 /// streamed using [`ObjectStore::put_multipart`]
 pub struct BufWriter {
     capacity: usize,
@@ -242,7 +242,7 @@ enum BufWriterState {
     Prepare(BoxFuture<'static, crate::Result<WriteMultipart>>),
     /// Write to a multipart upload
     Write(Option<WriteMultipart>),
-    /// [`ObjectStore::put`]
+    /// [`ObjectStore::put_opts`]
     Flush(BoxFuture<'static, crate::Result<()>>),
 }
 
@@ -489,7 +489,7 @@ mod tests {
     use super::*;
     use crate::memory::InMemory;
     use crate::path::Path;
-    use crate::{Attribute, GetOptions};
+    use crate::{Attribute, GetOptions, ObjectStoreExt};
     use itertools::Itertools;
     use tokio::io::{AsyncBufReadExt, AsyncReadExt, AsyncSeekExt, AsyncWriteExt};
 

src/chunked.rs

@@ -185,6 +185,7 @@ mod tests {
     use crate::local::LocalFileSystem;
     use crate::memory::InMemory;
     use crate::path::Path;
+    use crate::ObjectStoreExt;
 
     use super::*;
 

src/gcp/mod.rs

@@ -287,6 +287,7 @@ mod test {
 
     use crate::integration::*;
     use crate::tests::*;
+    use crate::ObjectStoreExt;
 
     use super::*;
 

src/integration.rs

@@ -29,7 +29,7 @@ use crate::multipart::MultipartStore;
 use crate::path::Path;
 use crate::{
     Attribute, Attributes, DynObjectStore, Error, GetOptions, GetRange, MultipartUpload,
-    ObjectStore, PutMode, PutPayload, UpdateVersion, WriteMultipart,
+    ObjectStore, ObjectStoreExt, PutMode, PutPayload, UpdateVersion, WriteMultipart,
 };
 use bytes::Bytes;
 use futures::stream::FuturesUnordered;

src/lib.rs

@@ -252,11 +252,11 @@
 //!
 //! # Put Object
 //!
-//! Use the [`ObjectStore::put`] method to atomically write data.
+//! Use the [`ObjectStoreExt::put`] method to atomically write data.
 //!
 //! ```
 //! # use object_store::local::LocalFileSystem;
-//! # use object_store::{ObjectStore, PutPayload};
+//! # use object_store::{ObjectStore, ObjectStoreExt, PutPayload};
 //! # use std::sync::Arc;
 //! # use object_store::path::Path;
 //! # fn get_object_store() -> Arc<dyn ObjectStore> {
@@ -338,7 +338,7 @@
 //!
 //! ```
 //! # use object_store::local::LocalFileSystem;
-//! # use object_store::{ObjectStore, PutPayloadMut};
+//! # use object_store::{ObjectStore, ObjectStoreExt, PutPayloadMut};
 //! # use std::sync::Arc;
 //! # use bytes::Bytes;
 //! # use tokio::io::AsyncWriteExt;
@@ -587,19 +587,22 @@ pub type DynObjectStore = dyn ObjectStore;
 pub type MultipartId = String;
 
 /// Universal API to multiple object store services.
+///
+/// For more convience methods, check [`ObjectStoreExt`].
+///
+/// # Contract
+/// This trait is meant as a contract between object store implementations
+/// (e.g. providers, wrappers) and the `object_store` crate itself.
+///
+/// The [`ObjectStoreExt`] acts as an API/contract between `object_store`
+/// and the store users.
 #[async_trait]
 pub trait ObjectStore: std::fmt::Display + Send + Sync + Debug + 'static {
-    /// Save the provided bytes to the specified location
+    /// Save the provided `payload` to `location` with the given options
     ///
     /// The operation is guaranteed to be atomic, it will either successfully
     /// write the entirety of `payload` to `location`, or fail. No clients
     /// should be able to observe a partially written object
-    async fn put(&self, location: &Path, payload: PutPayload) -> Result<PutResult> {
-        self.put_opts(location, payload, PutOptions::default())
-            .await
-    }
-
-    /// Save the provided `payload` to `location` with the given options
     async fn put_opts(
         &self,
         location: &Path,
@@ -609,7 +612,7 @@ pub trait ObjectStore: std::fmt::Display + Send + Sync + Debug + 'static {
 
     /// Perform a multipart upload
     ///
-    /// Client should prefer [`ObjectStore::put`] for small payloads, as streaming uploads
+    /// Client should prefer [`ObjectStoreExt::put`] for small payloads, as streaming uploads
     /// typically require multiple separate requests. See [`MultipartUpload`] for more information
     ///
     /// For more advanced multipart uploads see [`MultipartStore`](multipart::MultipartStore)
@@ -620,7 +623,7 @@ pub trait ObjectStore: std::fmt::Display + Send + Sync + Debug + 'static {
 
     /// Perform a multipart upload with options
     ///
-    /// Client should prefer [`ObjectStore::put`] for small payloads, as streaming uploads
+    /// Client should prefer [`ObjectStore::put_opts`] for small payloads, as streaming uploads
     /// typically require multiple separate requests. See [`MultipartUpload`] for more information
     ///
     /// For more advanced multipart uploads see [`MultipartStore`](multipart::MultipartStore)
@@ -696,7 +699,7 @@ pub trait ObjectStore: std::fmt::Display + Send + Sync + Debug + 'static {
     /// # async fn example() -> Result<(), Box<dyn std::error::Error>> {
     /// # let root = tempfile::TempDir::new().unwrap();
     /// # let store = LocalFileSystem::new_with_prefix(root.path()).unwrap();
-    /// # use object_store::{ObjectStore, ObjectMeta};
+    /// # use object_store::{ObjectStore, ObjectStoreExt, ObjectMeta};
     /// # use object_store::path::Path;
     /// # use futures::{StreamExt, TryStreamExt};
     /// #
@@ -803,10 +806,6 @@ macro_rules! as_ref_impl {
     ($type:ty) => {
         #[async_trait]
         impl ObjectStore for $type {
-            async fn put(&self, location: &Path, payload: PutPayload) -> Result<PutResult> {
-                self.as_ref().put(location, payload).await
-            }
-
             async fn put_opts(
                 &self,
                 location: &Path,
@@ -901,6 +900,40 @@ macro_rules! as_ref_impl {
 as_ref_impl!(Arc<dyn ObjectStore>);
 as_ref_impl!(Box<dyn ObjectStore>);
 
+/// Helper module to [seal traits](https://predr.ag/blog/definitive-guide-to-sealed-traits-in-rust/).
+mod private {
+    pub trait Sealed {}
+
+    impl<T> Sealed for T where T: super::ObjectStore + ?Sized {}
+}
+
+/// Extension trait for [`ObjectStore`] with convinience functions.
+///
+/// See "contract" section within the [`ObjectStore`] documentation for more reasoning.
+///
+/// # Implementation
+/// You MUST NOT implement this trait yourself. It is automatically implemented for all [`ObjectStore`] implementations.
+#[async_trait]
+pub trait ObjectStoreExt: private::Sealed {
+    /// Save the provided bytes to the specified location
+    ///
+    /// The operation is guaranteed to be atomic, it will either successfully
+    /// write the entirety of `payload` to `location`, or fail. No clients
+    /// should be able to observe a partially written object
+    async fn put(&self, location: &Path, payload: PutPayload) -> Result<PutResult>;
+}
+
+#[async_trait]
+impl<T> ObjectStoreExt for T
+where
+    T: ObjectStore + private::Sealed + ?Sized,
+{
+    async fn put(&self, location: &Path, payload: PutPayload) -> Result<PutResult> {
+        self.put_opts(location, payload, PutOptions::default())
+            .await
+    }
+}
+
 /// Result of a list call that includes objects, prefixes (directories) and a
 /// token for the next set of results. Individual result sets may be limited to
 /// 1,000 objects based on the underlying object storage's limitations.

src/limit.rs

@@ -71,11 +71,6 @@ impl<T: ObjectStore> std::fmt::Display for LimitStore<T> {
 
 #[async_trait]
 impl<T: ObjectStore> ObjectStore for LimitStore<T> {
-    async fn put(&self, location: &Path, payload: PutPayload) -> Result<PutResult> {
-        let _permit = self.semaphore.acquire().await.unwrap();
-        self.inner.put(location, payload).await
-    }
-
     async fn put_opts(
         &self,
         location: &Path,

src/local.rs

@@ -1090,7 +1090,7 @@ mod tests {
     #[cfg(target_family = "unix")]
     use tempfile::NamedTempFile;
 
-    use crate::integration::*;
+    use crate::{integration::*, ObjectStoreExt};
 
     use super::*;
 

src/memory.rs

@@ -536,7 +536,7 @@ impl MultipartUpload for InMemoryUpload {
 
 #[cfg(test)]
 mod tests {
-    use crate::integration::*;
+    use crate::{integration::*, ObjectStoreExt};
 
     use super::*;
 

src/prefix.rs

@@ -98,11 +98,6 @@ fn strip_meta(prefix: &Path, meta: ObjectMeta) -> ObjectMeta {
 }
 #[async_trait::async_trait]
 impl<T: ObjectStore> ObjectStore for PrefixStore<T> {
-    async fn put(&self, location: &Path, payload: PutPayload) -> Result<PutResult> {
-        let full_path = self.full_path(location);
-        self.inner.put(&full_path, payload).await
-    }
-
     async fn put_opts(
         &self,
         location: &Path,
@@ -225,8 +220,8 @@ impl<T: ObjectStore> ObjectStore for PrefixStore<T> {
 #[cfg(test)]
 mod tests {
     use super::*;
-    use crate::integration::*;
     use crate::local::LocalFileSystem;
+    use crate::{integration::*, ObjectStoreExt};
 
     use tempfile::TempDir;
 

src/throttle.rs

@@ -90,7 +90,7 @@ pub struct ThrottleConfig {
     /// [`wait_list_with_delimiter_per_call`](Self::wait_list_with_delimiter_per_call).
     pub wait_list_with_delimiter_per_entry: Duration,
 
-    /// Sleep duration for every call to [`put`](ThrottledStore::put).
+    /// Sleep duration for every call to [`put_opts`](ThrottledStore::put_opts).
     ///
     /// Sleeping is done before the underlying store is called and independently of the success of
     /// the operation.
@@ -148,11 +148,6 @@ impl<T: ObjectStore> std::fmt::Display for ThrottledStore<T> {
 
 #[async_trait]
 impl<T: ObjectStore> ObjectStore for ThrottledStore<T> {
-    async fn put(&self, location: &Path, payload: PutPayload) -> Result<PutResult> {
-        sleep(self.config().wait_put_per_call).await;
-        self.inner.put(location, payload).await
-    }
-
     async fn put_opts(
         &self,
         location: &Path,
@@ -404,6 +399,7 @@ impl MultipartUpload for ThrottledUpload {
 #[cfg(test)]
 mod tests {
     use super::*;
+    use crate::ObjectStoreExt;
     use crate::{integration::*, memory::InMemory, GetResultPayload};
     use futures::TryStreamExt;
     use tokio::time::Duration;