opendal_service_lakefs/
backend.rs

1// Licensed to the Apache Software Foundation (ASF) under one
2// or more contributor license agreements.  See the NOTICE file
3// distributed with this work for additional information
4// regarding copyright ownership.  The ASF licenses this file
5// to you under the Apache License, Version 2.0 (the
6// "License"); you may not use this file except in compliance
7// with the License.  You may obtain a copy of the License at
8//
9//   http://www.apache.org/licenses/LICENSE-2.0
10//
11// Unless required by applicable law or agreed to in writing,
12// software distributed under the License is distributed on an
13// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
14// KIND, either express or implied.  See the License for the
15// specific language governing permissions and limitations
16// under the License.
17
18use std::sync::Arc;
19
20use bytes::Buf;
21use http::StatusCode;
22use log::debug;
23use opendal_core::raw::*;
24use opendal_core::*;
25
26use super::LAKEFS_SCHEME;
27use super::config::LakefsConfig;
28use super::core::LakefsStatus;
29use super::core::parse_error;
30use super::core::{ErrorContext, LakefsCore};
31use super::deleter::LakefsDeleter;
32use super::lister::LakefsLister;
33use super::reader::*;
34use super::writer::LakefsWriter;
35
36/// [Lakefs](https://docs.lakefs.io/reference/api.html#/)'s API support.
37#[doc = include_str!("docs.md")]
38#[derive(Debug, Default)]
39pub struct LakefsBuilder {
40    pub(super) config: LakefsConfig,
41}
42
43impl LakefsBuilder {
44    /// Set the endpoint of this backend.
45    ///
46    /// endpoint must be full uri.
47    ///
48    /// This is required.
49    /// - `http://127.0.0.1:8000` (lakefs daemon in local)
50    /// - `https://my-lakefs.example.com` (lakefs server)
51    pub fn endpoint(mut self, endpoint: &str) -> Self {
52        if !endpoint.is_empty() {
53            self.config.endpoint = Some(endpoint.to_string());
54        }
55        self
56    }
57
58    /// Set username of this backend. This is required.
59    pub fn username(mut self, username: &str) -> Self {
60        if !username.is_empty() {
61            self.config.username = Some(username.to_string());
62        }
63        self
64    }
65
66    /// Set password of this backend. This is required.
67    pub fn password(mut self, password: &str) -> Self {
68        if !password.is_empty() {
69            self.config.password = Some(password.to_string());
70        }
71        self
72    }
73
74    /// Set branch of this backend or a commit ID. Default is main.
75    ///
76    /// Branch can be a branch name.
77    ///
78    /// For example, branch can be:
79    /// - main
80    /// - 1d0c4eb
81    pub fn branch(mut self, branch: &str) -> Self {
82        if !branch.is_empty() {
83            self.config.branch = Some(branch.to_string());
84        }
85        self
86    }
87
88    /// Set root of this backend.
89    ///
90    /// All operations will happen under this root.
91    pub fn root(mut self, root: &str) -> Self {
92        if !root.is_empty() {
93            self.config.root = Some(root.to_string());
94        }
95        self
96    }
97
98    /// Set the repository of this backend.
99    ///
100    /// This is required.
101    pub fn repository(mut self, repository: &str) -> Self {
102        if !repository.is_empty() {
103            self.config.repository = Some(repository.to_string());
104        }
105        self
106    }
107}
108
109impl Builder for LakefsBuilder {
110    type Config = LakefsConfig;
111
112    /// Build a LakefsBackend.
113    fn build(self) -> Result<impl Service> {
114        debug!("backend build started: {:?}", self);
115
116        let endpoint = match self.config.endpoint {
117            Some(endpoint) => Ok(endpoint.clone()),
118            None => Err(Error::new(ErrorKind::ConfigInvalid, "endpoint is empty")
119                .with_operation("Builder::build")
120                .with_context("service", LAKEFS_SCHEME)),
121        }?;
122        debug!("backend use endpoint: {:?}", endpoint);
123
124        let repository = match &self.config.repository {
125            Some(repository) => Ok(repository.clone()),
126            None => Err(Error::new(ErrorKind::ConfigInvalid, "repository is empty")
127                .with_operation("Builder::build")
128                .with_context("service", LAKEFS_SCHEME)),
129        }?;
130        debug!("backend use repository: {}", repository);
131
132        let branch = match &self.config.branch {
133            Some(branch) => branch.clone(),
134            None => "main".to_string(),
135        };
136        debug!("backend use branch: {}", branch);
137
138        let root = normalize_root(&self.config.root.unwrap_or_default());
139        debug!("backend use root: {}", root);
140
141        let username = match &self.config.username {
142            Some(username) => Ok(username.clone()),
143            None => Err(Error::new(ErrorKind::ConfigInvalid, "username is empty")
144                .with_operation("Builder::build")
145                .with_context("service", LAKEFS_SCHEME)),
146        }?;
147
148        let password = match &self.config.password {
149            Some(password) => Ok(password.clone()),
150            None => Err(Error::new(ErrorKind::ConfigInvalid, "password is empty")
151                .with_operation("Builder::build")
152                .with_context("service", LAKEFS_SCHEME)),
153        }?;
154
155        Ok(LakefsBackend {
156            core: Arc::new(LakefsCore {
157                info: ServiceInfo::new(LAKEFS_SCHEME, "", ""),
158                capability: Capability {
159                    stat: true,
160
161                    list: true,
162
163                    read: true,
164                    read_with_suffix: true,
165                    write: true,
166                    delete: true,
167                    copy: true,
168                    shared: true,
169                    ..Default::default()
170                },
171                endpoint,
172                repository,
173                branch,
174                root,
175                username,
176                password,
177            }),
178        })
179    }
180}
181
182/// Backend for Lakefs service
183#[derive(Debug, Clone)]
184pub struct LakefsBackend {
185    pub(crate) core: Arc<LakefsCore>,
186}
187
188impl Service for LakefsBackend {
189    type Reader = oio::StreamReader<LakefsReader>;
190    type Writer = oio::OneShotWriter<LakefsWriter>;
191    type Lister = oio::PageLister<LakefsLister>;
192    type Deleter = oio::OneShotDeleter<LakefsDeleter>;
193    type Copier = oio::OneShotCopier;
194    type Composer = ();
195
196    fn info(&self) -> ServiceInfo {
197        self.core.info.clone()
198    }
199
200    fn capability(&self) -> Capability {
201        self.core.capability
202    }
203
204    async fn create_dir(
205        &self,
206        _ctx: &OperationContext,
207        _path: &str,
208        _args: OpCreateDir,
209    ) -> Result<RpCreateDir> {
210        Err(Error::new(
211            ErrorKind::Unsupported,
212            "operation is not supported",
213        ))
214    }
215
216    async fn stat(&self, ctx: &OperationContext, path: &str, _: OpStat) -> Result<RpStat> {
217        // Stat root always returns a DIR.
218        if path == "/" {
219            return Ok(RpStat::new(MetadataBuilder::dir().build()));
220        }
221
222        let resp = self.core.get_object_metadata(ctx, path).await?;
223
224        let status = resp.status();
225
226        match status {
227            StatusCode::OK => {
228                let bs = resp.into_body();
229
230                let decoded_response: LakefsStatus =
231                    serde_json::from_reader(bs.reader()).map_err(new_json_deserialize_error)?;
232
233                // Use the helper function to parse LakefsStatus into Metadata
234                let meta = LakefsCore::parse_lakefs_status_into_metadata(&decoded_response)?;
235
236                Ok(RpStat::new(meta))
237            }
238            _ => Err(parse_error(
239                ErrorContext::new(ServiceOperation("StatObject")),
240                resp,
241            )),
242        }
243    }
244    fn read(&self, ctx: &OperationContext, path: &str, args: OpRead) -> Result<Self::Reader> {
245        let output: oio::StreamReader<LakefsReader> = {
246            Ok(oio::StreamReader::new(LakefsReader::new(
247                self.clone(),
248                ctx.clone(),
249                path,
250                args,
251            )))
252        }?;
253
254        Ok(output)
255    }
256
257    fn list(&self, ctx: &OperationContext, path: &str, args: OpList) -> Result<Self::Lister> {
258        let output: oio::PageLister<LakefsLister> = {
259            let l = LakefsLister::new(
260                self.core.clone(),
261                ctx.clone(),
262                path.to_string(),
263                args.limit(),
264                args.start_after(),
265                args.recursive(),
266            );
267
268            Ok(oio::PageLister::new(l))
269        }?;
270
271        Ok(output)
272    }
273
274    fn write(&self, ctx: &OperationContext, path: &str, args: OpWrite) -> Result<Self::Writer> {
275        let output: oio::OneShotWriter<LakefsWriter> = {
276            Ok(oio::OneShotWriter::new(LakefsWriter::new(
277                self.core.clone(),
278                ctx.clone(),
279                path.to_string(),
280                args,
281            )))
282        }?;
283
284        Ok(output)
285    }
286
287    fn delete(&self, ctx: &OperationContext) -> Result<Self::Deleter> {
288        let output: oio::OneShotDeleter<LakefsDeleter> = {
289            Ok(oio::OneShotDeleter::new(LakefsDeleter::new(
290                self.core.clone(),
291                ctx.clone(),
292            )))
293        }?;
294
295        Ok(output)
296    }
297
298    fn copy(
299        &self,
300        ctx: &OperationContext,
301        from: &str,
302        to: &str,
303        args: OpCopy,
304    ) -> Result<Self::Copier> {
305        let backend = self.clone();
306        let core = self.core.clone();
307        let ctx = ctx.clone();
308        let from = from.to_string();
309        let to = to.to_string();
310        let source_content_length_hint = args.source_content_length_hint();
311
312        Ok(oio::OneShotCopier::new(async move {
313            let source_size = match source_content_length_hint {
314                Some(size) => size,
315                None => backend
316                    .stat(&ctx, &from, OpStat::default())
317                    .await?
318                    .into_metadata()
319                    .content_length(),
320            };
321
322            let resp = core.copy_object(&ctx, &from, &to).await?;
323            let status = resp.status();
324
325            match status {
326                StatusCode::CREATED => Ok(MetadataBuilder::file(source_size).build()),
327                _ => Err(parse_error(
328                    ErrorContext::new(ServiceOperation("CopyObject")),
329                    resp,
330                )),
331            }
332        }))
333    }
334
335    async fn rename(
336        &self,
337        _ctx: &OperationContext,
338        _from: &str,
339        _to: &str,
340        _args: OpRename,
341    ) -> Result<RpRename> {
342        Err(Error::new(
343            ErrorKind::Unsupported,
344            "operation is not supported",
345        ))
346    }
347
348    async fn presign(
349        &self,
350        _ctx: &OperationContext,
351        _path: &str,
352        _args: OpPresign,
353    ) -> Result<RpPresign> {
354        Err(Error::new(
355            ErrorKind::Unsupported,
356            "operation is not supported",
357        ))
358    }
359}
opendal_service_lakefs/backend.rs

opendal_service_lakefs/
backend.rs
