1

core/src/services/lakefs/backend.rs

@@ -30,6 +30,7 @@ use super::core::LakefsStatus;
 use super::error::parse_error;
 use super::lister::LakefsLister;
 use crate::raw::*;
+use crate::services::lakefs::writer::LakefsWriter;
 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

@@ -17,9 +17,10 @@
 
 use std::fmt::Debug;
 
-use http::header;
+use http::header::{CACHE_CONTROL, CONTENT_DISPOSITION, CONTENT_LENGTH, CONTENT_TYPE};
 use http::Request;
 use http::Response;
+use http::{header, HeaderName};
 use serde::Deserialize;
 
 use crate::raw::*;
@@ -143,6 +144,51 @@ impl LakefsCore {
 
  self.client.send(req).await
  }
+
+ pub async fn upload_object(
+ &self,
+ path: &str,
+ args: &OpWrite,
+ body: Buffer,
+ ) -> Result<Request<Buffer>> {
+ let p = build_abs_path(&self.root, path)
+ .trim_end_matches('/')
+ .to_string();
+
+ let url = format!(
+ "{}/api/v1/repositories/{}/refs/{}/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);
+
+ if let Some(mime) = args.content_type() {
+ req = req.header(CONTENT_TYPE, mime)
+ }
+
+ if let Some(pos) = args.content_disposition() {
+ req = req.header(CONTENT_DISPOSITION, pos)
+ }
+
+ if let Some(cache_control) = args.cache_control() {
+ req = req.header(CACHE_CONTROL, cache_control)
+ }
+
+ let req = req.body(body).map_err(new_request_build_error)?;
+
+ Ok(req)
+ }
+
+ #[inline]
+ pub async fn send(&self, req: Request<Buffer>) -> Result<Response<Buffer>> {
+ self.client.send(req).await
+ }
 }
 
 #[derive(Deserialize, Eq, PartialEq, Debug)]

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,53 @@
+// 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 req = self.core.upload_object(&self.path, &self.op, bs).await?;
+
+ let resp = self.core.send(req).await?;
+
+ let status = resp.status();
+
+ match status {
+ StatusCode::CREATED | StatusCode::OK => Ok(()),
+ _ => Err(parse_error(resp).await?),
+ }
+ }
+}