opendal_service_b2/
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::fmt::Debug;
19use std::sync::Arc;
20
21use http::Request;
22use http::StatusCode;
23use log::debug;
24use mea::rwlock::RwLock;
25
26use super::B2_SCHEME;
27use super::config::B2Config;
28use super::core::B2Core;
29use super::core::B2Signer;
30use super::core::constants;
31use super::core::parse_error;
32use super::core::parse_file_info;
33use super::deleter::B2Deleter;
34use super::lister::B2Lister;
35use super::reader::*;
36use super::writer::B2Writer;
37use super::writer::B2Writers;
38use opendal_core::raw::*;
39use opendal_core::*;
40
41/// [b2](https://www.backblaze.com/cloud-storage) services support.
42#[doc = include_str!("docs.md")]
43#[derive(Default)]
44pub struct B2Builder {
45    pub(super) config: B2Config,
46}
47
48impl Debug for B2Builder {
49    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
50        f.debug_struct("B2Builder")
51            .field("config", &self.config)
52            .finish_non_exhaustive()
53    }
54}
55
56impl B2Builder {
57    /// Set root of this backend.
58    ///
59    /// All operations will happen under this root.
60    pub fn root(mut self, root: &str) -> Self {
61        self.config.root = if root.is_empty() {
62            None
63        } else {
64            Some(root.to_string())
65        };
66
67        self
68    }
69
70    /// application_key_id of this backend.
71    pub fn application_key_id(mut self, application_key_id: &str) -> Self {
72        self.config.application_key_id = if application_key_id.is_empty() {
73            None
74        } else {
75            Some(application_key_id.to_string())
76        };
77
78        self
79    }
80
81    /// application_key of this backend.
82    pub fn application_key(mut self, application_key: &str) -> Self {
83        self.config.application_key = if application_key.is_empty() {
84            None
85        } else {
86            Some(application_key.to_string())
87        };
88
89        self
90    }
91
92    /// Set bucket name of this backend.
93    /// You can find it in <https://secure.backblaze.com/b2_buckets.html>
94    pub fn bucket(mut self, bucket: &str) -> Self {
95        self.config.bucket = bucket.to_string();
96
97        self
98    }
99
100    /// Set bucket id of this backend.
101    /// You can find it in <https://secure.backblaze.com/b2_buckets.html>
102    pub fn bucket_id(mut self, bucket_id: &str) -> Self {
103        self.config.bucket_id = bucket_id.to_string();
104
105        self
106    }
107}
108
109impl Builder for B2Builder {
110    type Config = B2Config;
111
112    /// Builds the backend and returns the result of B2Backend.
113    fn build(self) -> Result<impl Service> {
114        debug!("backend build started: {:?}", self);
115
116        let root = normalize_root(&self.config.root.clone().unwrap_or_default());
117        debug!("backend use root {}", root);
118
119        // Handle bucket.
120        if self.config.bucket.is_empty() {
121            return Err(Error::new(ErrorKind::ConfigInvalid, "bucket is empty")
122                .with_operation("Builder::build")
123                .with_context("service", B2_SCHEME));
124        }
125
126        debug!("backend use bucket {}", self.config.bucket);
127
128        // Handle bucket_id.
129        if self.config.bucket_id.is_empty() {
130            return Err(Error::new(ErrorKind::ConfigInvalid, "bucket_id is empty")
131                .with_operation("Builder::build")
132                .with_context("service", B2_SCHEME));
133        }
134
135        debug!("backend bucket_id {}", self.config.bucket_id);
136
137        let application_key_id = match &self.config.application_key_id {
138            Some(application_key_id) => Ok(application_key_id.clone()),
139            None => Err(
140                Error::new(ErrorKind::ConfigInvalid, "application_key_id is empty")
141                    .with_operation("Builder::build")
142                    .with_context("service", B2_SCHEME),
143            ),
144        }?;
145
146        let application_key = match &self.config.application_key {
147            Some(key_id) => Ok(key_id.clone()),
148            None => Err(
149                Error::new(ErrorKind::ConfigInvalid, "application_key is empty")
150                    .with_operation("Builder::build")
151                    .with_context("service", B2_SCHEME),
152            ),
153        }?;
154
155        let signer = B2Signer {
156            application_key_id,
157            application_key,
158            ..Default::default()
159        };
160
161        Ok(B2Backend {
162            core: Arc::new(B2Core {
163                info: ServiceInfo::new(B2_SCHEME, &root, ""),
164                capability: Capability {
165                    stat: true,
166
167                    read: true,
168                    read_with_suffix: true,
169
170                    write: true,
171                    write_can_empty: true,
172                    write_can_multi: true,
173                    write_with_content_type: true,
174                    write_with_user_metadata: true,
175                    // The min multipart size of b2 is 5 MiB.
176                    //
177                    // ref: <https://www.backblaze.com/docs/cloud-storage-large-files>
178                    write_multi_min_size: Some(5 * 1024 * 1024),
179                    // The max multipart size of b2 is 5 Gb.
180                    //
181                    // ref: <https://www.backblaze.com/docs/cloud-storage-large-files>
182                    write_multi_max_size: if cfg!(target_pointer_width = "64") {
183                        Some(5 * 1024 * 1024 * 1024)
184                    } else {
185                        Some(usize::MAX)
186                    },
187
188                    delete: true,
189                    copy: true,
190
191                    list: true,
192                    list_with_limit: true,
193                    list_with_start_after: true,
194                    list_with_recursive: true,
195
196                    presign: true,
197                    presign_read: true,
198                    presign_write: true,
199                    presign_stat: true,
200
201                    shared: true,
202
203                    ..Default::default()
204                },
205                signer: Arc::new(RwLock::new(signer)),
206                root,
207
208                bucket: self.config.bucket.clone(),
209                bucket_id: self.config.bucket_id.clone(),
210            }),
211        })
212    }
213}
214
215/// Backend for b2 services.
216#[derive(Debug, Clone)]
217pub struct B2Backend {
218    pub(crate) core: Arc<B2Core>,
219}
220
221impl Service for B2Backend {
222    type Reader = oio::StreamReader<B2Reader>;
223    type Writer = B2Writers;
224    type Lister = oio::PageLister<B2Lister>;
225    type Deleter = oio::OneShotDeleter<B2Deleter>;
226    type Copier = oio::OneShotCopier;
227
228    fn info(&self) -> ServiceInfo {
229        self.core.info.clone()
230    }
231
232    fn capability(&self) -> Capability {
233        self.core.capability
234    }
235
236    async fn create_dir(
237        &self,
238        _ctx: &OperationContext,
239        _path: &str,
240        _args: OpCreateDir,
241    ) -> Result<RpCreateDir> {
242        Err(Error::new(
243            ErrorKind::Unsupported,
244            "operation is not supported",
245        ))
246    }
247
248    /// B2 have a get_file_info api required a file_id field, but field_id need call list api, list api also return file info
249    /// So we call list api to get file info
250    async fn stat(&self, ctx: &OperationContext, path: &str, _args: OpStat) -> Result<RpStat> {
251        // Stat root always returns a DIR.
252        if path == "/" {
253            return Ok(RpStat::new(Metadata::new(EntryMode::DIR)));
254        }
255
256        let delimiter = if path.ends_with('/') { Some("/") } else { None };
257
258        let file_info = self.core.get_file_info(ctx, path, delimiter).await?;
259        let meta = parse_file_info(&file_info);
260        Ok(RpStat::new(meta))
261    }
262    fn read(&self, ctx: &OperationContext, path: &str, args: OpRead) -> Result<Self::Reader> {
263        let output: oio::StreamReader<B2Reader> = {
264            Ok(oio::StreamReader::new(B2Reader::new(
265                self.clone(),
266                ctx.clone(),
267                path,
268                args,
269            )))
270        }?;
271
272        Ok(output)
273    }
274
275    fn write(&self, ctx: &OperationContext, path: &str, args: OpWrite) -> Result<Self::Writer> {
276        let output: B2Writers = {
277            let concurrent = args.concurrent();
278            let writer = B2Writer::new(self.core.clone(), ctx.clone(), path, args);
279
280            let w = oio::MultipartWriter::new(ctx.executor().clone(), writer, concurrent);
281
282            Ok(w)
283        }?;
284
285        Ok(output)
286    }
287
288    fn delete(&self, ctx: &OperationContext) -> Result<Self::Deleter> {
289        let output: oio::OneShotDeleter<B2Deleter> = {
290            Ok(oio::OneShotDeleter::new(B2Deleter::new(
291                self.core.clone(),
292                ctx.clone(),
293            )))
294        }?;
295
296        Ok(output)
297    }
298
299    fn list(&self, ctx: &OperationContext, path: &str, args: OpList) -> Result<Self::Lister> {
300        let output: oio::PageLister<B2Lister> = {
301            Ok(oio::PageLister::new(B2Lister::new(
302                self.core.clone(),
303                ctx.clone(),
304                path,
305                args.recursive(),
306                args.limit(),
307                args.start_after(),
308            )))
309        }?;
310
311        Ok(output)
312    }
313
314    fn copy(
315        &self,
316        ctx: &OperationContext,
317        from: &str,
318        to: &str,
319        _args: OpCopy,
320        _opts: OpCopier,
321    ) -> Result<Self::Copier> {
322        let core = self.core.clone();
323        let ctx = ctx.clone();
324        let from = from.to_string();
325        let to = to.to_string();
326
327        Ok(oio::OneShotCopier::new(async move {
328            let file_info = core.get_file_info(&ctx, &from, None).await?;
329            let source_file_id = file_info.file_id;
330
331            let Some(source_file_id) = source_file_id else {
332                return Err(Error::new(ErrorKind::IsADirectory, "is a directory"));
333            };
334
335            let resp = core.copy_file(&ctx, source_file_id, &to).await?;
336
337            let status = resp.status();
338
339            match status {
340                StatusCode::OK => Ok(Metadata::default()),
341                _ => Err(parse_error(resp)),
342            }
343        }))
344    }
345
346    async fn rename(
347        &self,
348        _ctx: &OperationContext,
349        _from: &str,
350        _to: &str,
351        _args: OpRename,
352    ) -> Result<RpRename> {
353        Err(Error::new(
354            ErrorKind::Unsupported,
355            "operation is not supported",
356        ))
357    }
358
359    async fn presign(
360        &self,
361        ctx: &OperationContext,
362        path: &str,
363        args: OpPresign,
364    ) -> Result<RpPresign> {
365        match args.operation() {
366            PresignOperation::Stat(_) => {
367                let resp = self
368                    .core
369                    .get_download_authorization(ctx, path, args.expire())
370                    .await?;
371                let path = build_abs_path(&self.core.root, path);
372
373                let auth_info = self.core.get_auth_info(ctx).await?;
374
375                let url = format!(
376                    "{}/file/{}/{}?Authorization={}",
377                    auth_info.download_url, self.core.bucket, path, resp.authorization_token
378                );
379
380                let req = Request::get(url);
381
382                let req = req.body(Buffer::new()).map_err(new_request_build_error)?;
383
384                // We don't need this request anymore, consume
385                let (parts, _) = req.into_parts();
386
387                Ok(RpPresign::new(PresignedRequest::new(
388                    parts.method,
389                    parts.uri,
390                    parts.headers,
391                )))
392            }
393            PresignOperation::Read(_, _) => {
394                let resp = self
395                    .core
396                    .get_download_authorization(ctx, path, args.expire())
397                    .await?;
398                let path = build_abs_path(&self.core.root, path);
399
400                let auth_info = self.core.get_auth_info(ctx).await?;
401
402                let url = format!(
403                    "{}/file/{}/{}?Authorization={}",
404                    auth_info.download_url, self.core.bucket, path, resp.authorization_token
405                );
406
407                let req = Request::get(url);
408
409                let req = req.body(Buffer::new()).map_err(new_request_build_error)?;
410
411                // We don't need this request anymore, consume
412                let (parts, _) = req.into_parts();
413
414                Ok(RpPresign::new(PresignedRequest::new(
415                    parts.method,
416                    parts.uri,
417                    parts.headers,
418                )))
419            }
420            PresignOperation::Write(_) => {
421                let resp = self.core.get_upload_url(ctx).await?;
422
423                let mut req = Request::post(&resp.upload_url);
424
425                req = req.header(http::header::AUTHORIZATION, resp.authorization_token);
426                req = req.header("X-Bz-File-Name", build_abs_path(&self.core.root, path));
427                req = req.header(http::header::CONTENT_TYPE, "b2/x-auto");
428                req = req.header(constants::X_BZ_CONTENT_SHA1, "do_not_verify");
429
430                let req = req.body(Buffer::new()).map_err(new_request_build_error)?;
431                // We don't need this request anymore, consume it directly.
432                let (parts, _) = req.into_parts();
433
434                Ok(RpPresign::new(PresignedRequest::new(
435                    parts.method,
436                    parts.uri,
437                    parts.headers,
438                )))
439            }
440            PresignOperation::Delete(_) => Err(Error::new(
441                ErrorKind::Unsupported,
442                "operation is not supported",
443            )),
444            _ => Err(Error::new(
445                ErrorKind::Unsupported,
446                "operation is not supported",
447            )),
448        }
449    }
450}
opendal_service_b2/backend.rs

opendal_service_b2/
backend.rs
