feat: add write api for lakefs service. (#5100)

* 1

* 1

* 1

* 1

* 1

* 1

* 1

* 1

* 1

* 1

* 1

* 1

core/src/services/lakefs/backend.rs

@@ -29,6 +29,7 @@ use super::core::LakefsCore;
 use super::core::LakefsStatus;
 use super::error::parse_error;
 use super::lister::LakefsLister;
+use super::writer::LakefsWriter;
 use crate::raw::*;
 use crate::services::LakefsConfig;
 use crate::*;
@@ -193,7 +194,7 @@ pub struct LakefsBackend {
 
 impl Access for LakefsBackend {
     type Reader = HttpBody;
-    type Writer = ();
+    type Writer = oio::OneShotWriter<LakefsWriter>;
     type Lister = oio::PageLister<LakefsLister>;
     type BlockingReader = ();
     type BlockingWriter = ();
@@ -206,6 +207,7 @@ impl Access for LakefsBackend {
                 stat: true,
                 list: true,
                 read: true,
+                write: true,
 
                 ..Default::default()
             });
@@ -276,4 +278,11 @@ impl Access for LakefsBackend {
 
         Ok((RpList::default(), oio::PageLister::new(l)))
     }
+
+    async fn write(&self, path: &str, args: OpWrite) -> Result<(RpWrite, Self::Writer)> {
+        Ok((
+            RpWrite::default(),
+            oio::OneShotWriter::new(LakefsWriter::new(self.core.clone(), path.to_string(), args)),
+        ))
+    }
 }

core/src/services/lakefs/core.rs

@@ -143,6 +143,34 @@ impl LakefsCore {
 
         self.client.send(req).await
     }
+
+    pub async fn upload_object(
+        &self,
+        path: &str,
+        _args: &OpWrite,
+        body: Buffer,
+    ) -> Result<Response<Buffer>> {
+        let p = build_abs_path(&self.root, path)
+            .trim_end_matches('/')
+            .to_string();
+
+        let url = format!(
+            "{}/api/v1/repositories/{}/branches/{}/objects?path={}",
+            self.endpoint,
+            self.repository,
+            self.branch,
+            percent_encode_path(&p)
+        );
+
+        let mut req = Request::post(&url);
+
+        let auth_header_content = format_authorization_by_basic(&self.username, &self.password)?;
+        req = req.header(header::AUTHORIZATION, auth_header_content);
+
+        let req = req.body(body).map_err(new_request_build_error)?;
+
+        self.client.send(req).await
+    }
 }
 
 #[derive(Deserialize, Eq, PartialEq, Debug)]

core/src/services/lakefs/docs.md

@@ -9,7 +9,7 @@ This service can be used to:
 
 - [x] stat
 - [x] read
-- [ ] write
+- [x] write
 - [ ] create_dir
 - [ ] delete
 - [ ] copy

core/src/services/lakefs/mod.rs

@@ -23,10 +23,14 @@ mod error;
 #[cfg(feature = "services-lakefs")]
 mod lister;
 
+#[cfg(feature = "services-lakefs")]
+mod writer;
+
 #[cfg(feature = "services-lakefs")]
 mod backend;
 #[cfg(feature = "services-lakefs")]
 pub use backend::LakefsBuilder as Lakefs;
 
 mod config;
+
 pub use config::LakefsConfig;

core/src/services/lakefs/writer.rs

@@ -0,0 +1,51 @@
+// 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 std::sync::Arc;
+
+use http::StatusCode;
+
+use crate::raw::*;
+use crate::services::lakefs::core::LakefsCore;
+use crate::*;
+
+use super::error::parse_error;
+
+pub struct LakefsWriter {
+    core: Arc<LakefsCore>,
+    op: OpWrite,
+    path: String,
+}
+
+impl LakefsWriter {
+    pub fn new(core: Arc<LakefsCore>, path: String, op: OpWrite) -> Self {
+        LakefsWriter { core, path, op }
+    }
+}
+
+impl oio::OneShotWrite for LakefsWriter {
+    async fn write_once(&self, bs: Buffer) -> Result<()> {
+        let resp = self.core.upload_object(&self.path, &self.op, bs).await?;
+
+        let status = resp.status();
+
+        match status {
+            StatusCode::CREATED | StatusCode::OK => Ok(()),
+            _ => Err(parse_error(resp).await?),
+        }
+    }
+}