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::LakefsCore;
29use super::core::LakefsStatus;
30use super::core::parse_error;
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
195    fn info(&self) -> ServiceInfo {
196        self.core.info.clone()
197    }
198
199    fn capability(&self) -> Capability {
200        self.core.capability
201    }
202
203    async fn create_dir(
204        &self,
205        _ctx: &OperationContext,
206        _path: &str,
207        _args: OpCreateDir,
208    ) -> Result<RpCreateDir> {
209        Err(Error::new(
210            ErrorKind::Unsupported,
211            "operation is not supported",
212        ))
213    }
214
215    async fn stat(&self, ctx: &OperationContext, path: &str, _: OpStat) -> Result<RpStat> {
216        // Stat root always returns a DIR.
217        if path == "/" {
218            return Ok(RpStat::new(Metadata::new(EntryMode::DIR)));
219        }
220
221        let resp = self.core.get_object_metadata(ctx, path).await?;
222
223        let status = resp.status();
224
225        match status {
226            StatusCode::OK => {
227                let bs = resp.into_body();
228
229                let decoded_response: LakefsStatus =
230                    serde_json::from_reader(bs.reader()).map_err(new_json_deserialize_error)?;
231
232                // Use the helper function to parse LakefsStatus into Metadata
233                let meta = LakefsCore::parse_lakefs_status_into_metadata(&decoded_response);
234
235                Ok(RpStat::new(meta))
236            }
237            _ => Err(parse_error(resp)),
238        }
239    }
240    fn read(&self, ctx: &OperationContext, path: &str, args: OpRead) -> Result<Self::Reader> {
241        let output: oio::StreamReader<LakefsReader> = {
242            Ok(oio::StreamReader::new(LakefsReader::new(
243                self.clone(),
244                ctx.clone(),
245                path,
246                args,
247            )))
248        }?;
249
250        Ok(output)
251    }
252
253    fn list(&self, ctx: &OperationContext, path: &str, args: OpList) -> Result<Self::Lister> {
254        let output: oio::PageLister<LakefsLister> = {
255            let l = LakefsLister::new(
256                self.core.clone(),
257                ctx.clone(),
258                path.to_string(),
259                args.limit(),
260                args.start_after(),
261                args.recursive(),
262            );
263
264            Ok(oio::PageLister::new(l))
265        }?;
266
267        Ok(output)
268    }
269
270    fn write(&self, ctx: &OperationContext, path: &str, args: OpWrite) -> Result<Self::Writer> {
271        let output: oio::OneShotWriter<LakefsWriter> = {
272            Ok(oio::OneShotWriter::new(LakefsWriter::new(
273                self.core.clone(),
274                ctx.clone(),
275                path.to_string(),
276                args,
277            )))
278        }?;
279
280        Ok(output)
281    }
282
283    fn delete(&self, ctx: &OperationContext) -> Result<Self::Deleter> {
284        let output: oio::OneShotDeleter<LakefsDeleter> = {
285            Ok(oio::OneShotDeleter::new(LakefsDeleter::new(
286                self.core.clone(),
287                ctx.clone(),
288            )))
289        }?;
290
291        Ok(output)
292    }
293
294    fn copy(
295        &self,
296        ctx: &OperationContext,
297        from: &str,
298        to: &str,
299        _args: OpCopy,
300        _opts: OpCopier,
301    ) -> Result<Self::Copier> {
302        let core = self.core.clone();
303        let ctx = ctx.clone();
304        let from = from.to_string();
305        let to = to.to_string();
306
307        Ok(oio::OneShotCopier::new(async move {
308            let resp = core.copy_object(&ctx, &from, &to).await?;
309            let status = resp.status();
310
311            match status {
312                StatusCode::CREATED => Ok(Metadata::default()),
313                _ => Err(parse_error(resp)),
314            }
315        }))
316    }
317
318    async fn rename(
319        &self,
320        _ctx: &OperationContext,
321        _from: &str,
322        _to: &str,
323        _args: OpRename,
324    ) -> Result<RpRename> {
325        Err(Error::new(
326            ErrorKind::Unsupported,
327            "operation is not supported",
328        ))
329    }
330
331    async fn presign(
332        &self,
333        _ctx: &OperationContext,
334        _path: &str,
335        _args: OpPresign,
336    ) -> Result<RpPresign> {
337        Err(Error::new(
338            ErrorKind::Unsupported,
339            "operation is not supported",
340        ))
341    }
342}
opendal_service_lakefs/backend.rs

opendal_service_lakefs/
backend.rs
