feat(integrations/object_store): add AmazonS3Builder

core/src/types/scheme.rs

@@ -167,7 +167,7 @@ pub enum Scheme {
     Lakefs,
     /// [NebulaGraph](crate::services::NebulaGraph): NebulaGraph Services
     NebulaGraph,
-    /// Custom that allow users to implement services outside of OpenDAL.
+    /// Custom that allow users to implement services outside OpenDAL.
     ///
     /// # NOTE
     ///

integrations/object_store/Cargo.toml

@@ -29,6 +29,7 @@ version = "0.48.3"
 
 [features]
 send_wrapper = ["dep:send_wrapper"]
+services-s3 = ["opendal/services-s3"]
 
 [dependencies]
 async-trait = "0.1"

integrations/object_store/examples/basic.rs

@@ -3,24 +3,23 @@ use std::sync::Arc;
 use bytes::Bytes;
 use object_store::path::Path;
 use object_store::ObjectStore;
-use object_store_opendal::OpendalStore;
-use opendal::services::S3Config;
-use opendal::Operator;
 
+#[cfg(feature = "services-s3")]
+use object_store_opendal::AmazonS3Builder;
+
+#[cfg(feature = "services-s3")]
 #[tokio::main]
 async fn main() {
-    let mut cfg = S3Config::default();
-    cfg.access_key_id = Some("my_access_key".to_string());
-    cfg.secret_access_key = Some("my_secret_key".to_string());
-    cfg.endpoint = Some("my_endpoint".to_string());
-    cfg.region = Some("my_region".to_string());
-    cfg.bucket = "my_bucket".to_string();
-
-    // Create a new operator
-    let operator = Operator::from_config(cfg).unwrap().finish();
+    let s3_store = AmazonS3Builder::new()
+        .with_access_key_id("my_access_key")
+        .with_secret_access_key("my_secret_key")
+        .with_endpoint("my_endpoint")
+        .with_region("my_region")
+        .with_bucket_name("my_bucket")
+        .build()
+        .unwrap();
 
-    // Create a new object store
-    let object_store = Arc::new(OpendalStore::new(operator));
+    let object_store = Arc::new(s3_store);
 
     let path = Path::from("data/nested/test.txt");
     let bytes = Bytes::from_static(b"hello, world! I am nested.");
@@ -37,3 +36,8 @@ async fn main() {
 
     assert_eq!(content, bytes);
 }
+
+#[cfg(not(feature = "services-s3"))]
+fn main() {
+    println!("The 'services-s3' feature is not enabled.");
+}

integrations/object_store/src/lib.rs

@@ -65,6 +65,11 @@ pub use store::OpendalStore;
 
 mod utils;
 
+#[cfg(feature = "services-s3")]
+mod s3_builder;
+#[cfg(feature = "services-s3")]
+pub use s3_builder::AmazonS3Builder;
+
 // Make sure `send_wrapper` works as expected
 #[cfg(all(feature = "send_wrapper", target_arch = "wasm32"))]
 mod assert_send {

integrations/object_store/src/s3_builder.rs

@@ -0,0 +1,97 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use crate::utils::format_object_store_error;
+use crate::OpendalStore;
+use opendal::services::S3;
+use opendal::Operator;
+
+pub struct AmazonS3Builder {
+    builder: S3,
+}
+
+impl Default for AmazonS3Builder {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+impl AmazonS3Builder {
+    pub fn new() -> Self {
+        AmazonS3Builder {
+            builder: S3::default(),
+        }
+    }
+
+    pub fn root(mut self, root_path: impl Into<String>) -> Self {
+        self.builder = self.builder.root(root_path.into().as_str());
+        self
+    }
+
+    pub fn with_endpoint(mut self, endpoint: impl Into<String>) -> Self {
+        self.builder = self.builder.endpoint(endpoint.into().as_str());
+        self
+    }
+
+    pub fn with_region(mut self, region: impl Into<String>) -> Self {
+        self.builder = self.builder.region(region.into().as_str());
+        self
+    }
+
+    pub fn with_bucket_name(mut self, bucket_name: impl Into<String>) -> Self {
+        self.builder = self.builder.bucket(bucket_name.into().as_str());
+        self
+    }
+
+    pub fn with_access_key_id(mut self, access_key_id: impl Into<String>) -> Self {
+        self.builder = self.builder.access_key_id(access_key_id.into().as_str());
+        self
+    }
+
+    pub fn with_secret_access_key(mut self, secret_access_key: impl Into<String>) -> Self {
+        self.builder = self
+            .builder
+            .secret_access_key(secret_access_key.into().as_str());
+        self
+    }
+
+    pub fn with_token(mut self, token: impl Into<String>) -> Self {
+        self.builder = self.builder.session_token(token.into().as_str());
+        self
+    }
+
+    pub fn with_virtual_hosted_style_request(mut self, virtual_hosted_style_request: bool) -> Self {
+        if virtual_hosted_style_request {
+            self.builder = self.builder.enable_virtual_host_style();
+        }
+        self
+    }
+
+    pub fn with_skip_signature(mut self, skip_signature: bool) -> Self {
+        if skip_signature {
+            self.builder = self.builder.allow_anonymous();
+        }
+        self
+    }
+    pub fn build(self) -> object_store::Result<OpendalStore> {
+        let op = Operator::new(self.builder)
+            .map_err(|err| format_object_store_error(err, ""))?
+            .finish();
+
+        Ok(OpendalStore::new(op))
+    }
+}