opendal_service_dbfs/
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 serde::Deserialize;
24
25use super::DBFS_SCHEME;
26use super::config::DbfsConfig;
27use super::core::parse_error;
28use super::core::{DbfsCore, ErrorContext};
29use super::deleter::DbfsDeleter;
30use super::lister::DbfsLister;
31use super::writer::DbfsWriter;
32use opendal_core::raw::*;
33use opendal_core::*;
34
35/// [Dbfs](https://docs.databricks.com/api/azure/workspace/dbfs)'s REST API support.
36#[doc = include_str!("docs.md")]
37#[derive(Debug, Default)]
38pub struct DbfsBuilder {
39    pub(super) config: DbfsConfig,
40}
41
42impl DbfsBuilder {
43    /// Set root of this backend.
44    ///
45    /// All operations will happen under this root.
46    pub fn root(mut self, root: &str) -> Self {
47        self.config.root = if root.is_empty() {
48            None
49        } else {
50            Some(root.to_string())
51        };
52
53        self
54    }
55
56    /// Set endpoint of this backend.
57    ///
58    /// Endpoint must be full uri, e.g.
59    ///
60    /// - Azure: `https://adb-1234567890123456.78.azuredatabricks.net`
61    /// - Aws: `https://dbc-123a5678-90bc.cloud.databricks.com`
62    pub fn endpoint(mut self, endpoint: &str) -> Self {
63        self.config.endpoint = if endpoint.is_empty() {
64            None
65        } else {
66            Some(endpoint.trim_end_matches('/').to_string())
67        };
68        self
69    }
70
71    /// Set the token of this backend.
72    pub fn token(mut self, token: &str) -> Self {
73        if !token.is_empty() {
74            self.config.token = Some(token.to_string());
75        }
76        self
77    }
78}
79
80impl Builder for DbfsBuilder {
81    type Config = DbfsConfig;
82
83    /// Build a DbfsBackend.
84    fn build(self) -> Result<impl Service> {
85        debug!("backend build started: {:?}", self);
86
87        let root = normalize_root(&self.config.root.unwrap_or_default());
88        debug!("backend use root {root}");
89
90        let endpoint = match &self.config.endpoint {
91            Some(endpoint) => Ok(endpoint.clone()),
92            None => Err(Error::new(ErrorKind::ConfigInvalid, "endpoint is empty")
93                .with_operation("Builder::build")
94                .with_context("service", DBFS_SCHEME)),
95        }?;
96        debug!("backend use endpoint: {}", endpoint);
97
98        let token = match self.config.token {
99            Some(token) => token,
100            None => {
101                return Err(Error::new(
102                    ErrorKind::ConfigInvalid,
103                    "missing token for Dbfs",
104                ));
105            }
106        };
107
108        let capability = Capability {
109            stat: true,
110
111            write: true,
112            create_dir: true,
113            delete: true,
114            rename: true,
115
116            list: true,
117
118            shared: true,
119
120            ..Default::default()
121        };
122
123        Ok(DbfsBackend {
124            core: Arc::new(DbfsCore {
125                root,
126                endpoint: endpoint.to_string(),
127                token,
128            }),
129            capability,
130        })
131    }
132}
133
134/// Backend for DBFS service
135#[derive(Debug, Clone)]
136pub struct DbfsBackend {
137    core: Arc<DbfsCore>,
138    capability: Capability,
139}
140
141impl Service for DbfsBackend {
142    type Reader = ();
143    type Writer = oio::OneShotWriter<DbfsWriter>;
144    type Lister = oio::PageLister<DbfsLister>;
145    type Deleter = oio::OneShotDeleter<DbfsDeleter>;
146    type Copier = ();
147    type Composer = ();
148
149    fn info(&self) -> ServiceInfo {
150        ServiceInfo::new(DBFS_SCHEME, &self.core.root, "")
151    }
152
153    fn capability(&self) -> Capability {
154        self.capability
155    }
156
157    async fn create_dir(
158        &self,
159        ctx: &OperationContext,
160        path: &str,
161        _: OpCreateDir,
162    ) -> Result<RpCreateDir> {
163        let resp = self.core.dbfs_create_dir(ctx, path).await?;
164
165        let status = resp.status();
166
167        match status {
168            StatusCode::CREATED | StatusCode::OK => Ok(RpCreateDir::default()),
169            _ => Err(parse_error(
170                ErrorContext::new(ServiceOperation("Mkdirs")),
171                resp,
172            )),
173        }
174    }
175
176    async fn stat(&self, ctx: &OperationContext, path: &str, _: OpStat) -> Result<RpStat> {
177        // Stat root always returns a DIR.
178        if path == "/" {
179            return Ok(RpStat::new(MetadataBuilder::dir().build()));
180        }
181
182        let resp = self.core.dbfs_get_status(ctx, path).await?;
183
184        let status = resp.status();
185
186        match status {
187            StatusCode::OK => {
188                let mut meta = MetadataBuilder::unknown();
189                let bs = resp.into_body();
190                let decoded_response: DbfsStatus =
191                    serde_json::from_reader(bs.reader()).map_err(new_json_deserialize_error)?;
192                meta.last_modified(Timestamp::from_millisecond(
193                    decoded_response.modification_time,
194                )?);
195                match decoded_response.is_dir {
196                    true => meta.set_dir(),
197                    false => meta.set_file(decoded_response.file_size as u64),
198                };
199                Ok(RpStat::new(meta.build()))
200            }
201            StatusCode::NOT_FOUND if path.ends_with('/') => {
202                Ok(RpStat::new(MetadataBuilder::dir().build()))
203            }
204            _ => Err(parse_error(
205                ErrorContext::new(ServiceOperation("GetStatus")),
206                resp,
207            )),
208        }
209    }
210
211    fn read(&self, _ctx: &OperationContext, _path: &str, _args: OpRead) -> Result<Self::Reader> {
212        Err(Error::new(
213            ErrorKind::Unsupported,
214            "operation is not supported",
215        ))
216    }
217
218    fn write(&self, ctx: &OperationContext, path: &str, args: OpWrite) -> Result<Self::Writer> {
219        let output: oio::OneShotWriter<DbfsWriter> = {
220            Ok(oio::OneShotWriter::new(DbfsWriter::new(
221                self.core.clone(),
222                ctx.clone(),
223                args,
224                path.to_string(),
225            )))
226        }?;
227
228        Ok(output)
229    }
230
231    fn delete(&self, ctx: &OperationContext) -> Result<Self::Deleter> {
232        let output: oio::OneShotDeleter<DbfsDeleter> = {
233            Ok(oio::OneShotDeleter::new(DbfsDeleter::new(
234                self.core.clone(),
235                ctx.clone(),
236            )))
237        }?;
238
239        Ok(output)
240    }
241
242    fn list(&self, ctx: &OperationContext, path: &str, _args: OpList) -> Result<Self::Lister> {
243        let output: oio::PageLister<DbfsLister> = {
244            let l = DbfsLister::new(self.core.clone(), ctx.clone(), path.to_string());
245
246            Ok(oio::PageLister::new(l))
247        }?;
248
249        Ok(output)
250    }
251
252    async fn rename(
253        &self,
254        ctx: &OperationContext,
255        from: &str,
256        to: &str,
257        _args: OpRename,
258    ) -> Result<RpRename> {
259        self.core.dbfs_ensure_parent_path(ctx, to).await?;
260
261        let resp = self.core.dbfs_rename(ctx, from, to).await?;
262
263        let status = resp.status();
264
265        match status {
266            StatusCode::OK => Ok(RpRename::default()),
267            _ => Err(parse_error(
268                ErrorContext::new(ServiceOperation("Move")),
269                resp,
270            )),
271        }
272    }
273
274    fn copy(
275        &self,
276        _ctx: &OperationContext,
277        _from: &str,
278        _to: &str,
279        _args: OpCopy,
280    ) -> Result<Self::Copier> {
281        Err(Error::new(
282            ErrorKind::Unsupported,
283            "operation is not supported",
284        ))
285    }
286
287    async fn presign(
288        &self,
289        _ctx: &OperationContext,
290        _path: &str,
291        _args: OpPresign,
292    ) -> Result<RpPresign> {
293        Err(Error::new(
294            ErrorKind::Unsupported,
295            "operation is not supported",
296        ))
297    }
298}
299
300#[derive(Deserialize)]
301struct DbfsStatus {
302    // Not used fields.
303    // path: String,
304    is_dir: bool,
305    file_size: i64,
306    modification_time: i64,
307}
opendal_service_dbfs/backend.rs

opendal_service_dbfs/
backend.rs
