opendal_service_aliyun_drive/
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 asyncband::mutex::Mutex;
22use bytes::Buf;
23use log::debug;
24
25use super::ALIYUN_DRIVE_SCHEME;
26use super::config::AliyunDriveConfig;
27use super::core::*;
28use super::deleter::AliyunDriveDeleter;
29use super::lister::AliyunDriveLister;
30use super::reader::*;
31use super::writer::AliyunDriveLazyWriter;
32use opendal_core::raw::*;
33use opendal_core::*;
34
35#[doc = include_str!("docs.md")]
36#[derive(Default)]
37pub struct AliyunDriveBuilder {
38    pub(super) config: AliyunDriveConfig,
39}
40
41impl Debug for AliyunDriveBuilder {
42    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
43        f.debug_struct("AliyunDriveBuilder")
44            .field("config", &self.config)
45            .finish_non_exhaustive()
46    }
47}
48
49impl AliyunDriveBuilder {
50    /// Set the root of this backend.
51    ///
52    /// All operations will happen under this root.
53    pub fn root(mut self, root: &str) -> Self {
54        self.config.root = if root.is_empty() {
55            None
56        } else {
57            Some(root.to_string())
58        };
59
60        self
61    }
62
63    /// Set access_token of this backend.
64    pub fn access_token(mut self, access_token: &str) -> Self {
65        self.config.access_token = Some(access_token.to_string());
66
67        self
68    }
69
70    /// Set client_id of this backend.
71    pub fn client_id(mut self, client_id: &str) -> Self {
72        self.config.client_id = Some(client_id.to_string());
73
74        self
75    }
76
77    /// Set client_secret of this backend.
78    pub fn client_secret(mut self, client_secret: &str) -> Self {
79        self.config.client_secret = Some(client_secret.to_string());
80
81        self
82    }
83
84    /// Set refresh_token of this backend.
85    pub fn refresh_token(mut self, refresh_token: &str) -> Self {
86        self.config.refresh_token = Some(refresh_token.to_string());
87
88        self
89    }
90
91    /// Set drive_type of this backend.
92    pub fn drive_type(mut self, drive_type: &str) -> Self {
93        self.config.drive_type = drive_type.to_string();
94
95        self
96    }
97}
98
99impl Builder for AliyunDriveBuilder {
100    type Config = AliyunDriveConfig;
101
102    fn build(self) -> Result<impl Service> {
103        debug!("backend build started: {:?}", self);
104
105        let root = normalize_root(&self.config.root.clone().unwrap_or_default());
106        debug!("backend use root {}", root);
107
108        let sign = match self.config.access_token.clone() {
109            Some(access_token) if !access_token.is_empty() => {
110                AliyunDriveSign::Access(access_token)
111            }
112            _ => match (
113                self.config.client_id.clone(),
114                self.config.client_secret.clone(),
115                self.config.refresh_token.clone(),
116            ) {
117                (Some(client_id), Some(client_secret), Some(refresh_token)) if
118                !client_id.is_empty() && !client_secret.is_empty() && !refresh_token.is_empty() => {
119                    AliyunDriveSign::Refresh(client_id, client_secret, refresh_token, None, 0)
120                }
121                _ => return Err(Error::new(
122                    ErrorKind::ConfigInvalid,
123                    "access_token and a set of client_id, client_secret, and refresh_token are both missing.")
124                    .with_operation("Builder::build")
125                    .with_context("service", ALIYUN_DRIVE_SCHEME)),
126            },
127        };
128
129        let drive_type = match self.config.drive_type.as_str() {
130            "" | "default" => DriveType::Default,
131            "resource" => DriveType::Resource,
132            "backup" => DriveType::Backup,
133            _ => {
134                return Err(Error::new(
135                    ErrorKind::ConfigInvalid,
136                    "drive_type is invalid.",
137                ));
138            }
139        };
140        debug!("backend use drive_type {drive_type:?}");
141
142        Ok(AliyunDriveBackend {
143            core: Arc::new(AliyunDriveCore {
144                info: ServiceInfo::new(ALIYUN_DRIVE_SCHEME, &root, ""),
145                capability: Capability {
146                    stat: true,
147                    create_dir: true,
148                    read: true,
149                    read_with_suffix: true,
150                    write: true,
151                    write_can_multi: true,
152                    // The min multipart size of AliyunDrive is 100 KiB.
153                    write_multi_min_size: Some(100 * 1024),
154                    // The max multipart size of AliyunDrive is 5 GiB.
155                    write_multi_max_size: if cfg!(target_pointer_width = "64") {
156                        Some(5 * 1024 * 1024 * 1024)
157                    } else {
158                        Some(usize::MAX)
159                    },
160                    delete: true,
161                    copy: true,
162                    rename: true,
163                    list: true,
164                    list_with_limit: true,
165                    shared: true,
166                    ..Default::default()
167                },
168                endpoint: "https://openapi.alipan.com".to_string(),
169                root,
170                drive_type,
171                signer: Arc::new(Mutex::new(AliyunDriveSigner {
172                    drive_id: None,
173                    sign,
174                })),
175                dir_lock: Arc::new(Mutex::new(())),
176            }),
177        })
178    }
179}
180
181#[derive(Clone, Debug)]
182pub struct AliyunDriveBackend {
183    pub(crate) core: Arc<AliyunDriveCore>,
184}
185
186impl Service for AliyunDriveBackend {
187    type Reader = oio::StreamReader<AliyunDriveReader>;
188    type Writer = AliyunDriveLazyWriter;
189    type Lister = oio::PageLister<AliyunDriveLister>;
190    type Deleter = oio::OneShotDeleter<AliyunDriveDeleter>;
191    type Copier = oio::OneShotCopier;
192    type Composer = ();
193
194    fn info(&self) -> ServiceInfo {
195        self.core.info.clone()
196    }
197
198    fn capability(&self) -> Capability {
199        self.core.capability
200    }
201
202    async fn create_dir(
203        &self,
204        ctx: &OperationContext,
205        path: &str,
206        _args: OpCreateDir,
207    ) -> Result<RpCreateDir> {
208        self.core.ensure_dir_exists(ctx, path).await?;
209
210        Ok(RpCreateDir::default())
211    }
212
213    async fn rename(
214        &self,
215        ctx: &OperationContext,
216        from: &str,
217        to: &str,
218        _args: OpRename,
219    ) -> Result<RpRename> {
220        if from == to {
221            return Ok(RpRename::default());
222        }
223        let res = self.core.get_by_path(ctx, from).await?;
224        let file: AliyunDriveFile =
225            serde_json::from_reader(res.reader()).map_err(new_json_serialize_error)?;
226        // rename can overwrite.
227        match self.core.get_by_path(ctx, to).await {
228            Err(err) if err.kind() == ErrorKind::NotFound => {}
229            Err(err) => return Err(err),
230            Ok(res) => {
231                let file: AliyunDriveFile =
232                    serde_json::from_reader(res.reader()).map_err(new_json_serialize_error)?;
233                self.core.delete_path(ctx, &file.file_id).await?;
234            }
235        };
236
237        let parent_file_id = self.core.ensure_dir_exists(ctx, get_parent(to)).await?;
238        self.core
239            .move_path(ctx, &file.file_id, &parent_file_id)
240            .await?;
241
242        let from_name = get_basename(from);
243        let to_name = get_basename(to);
244
245        if from_name != to_name {
246            self.core.update_path(ctx, &file.file_id, to_name).await?;
247        }
248
249        Ok(RpRename::default())
250    }
251
252    fn copy(
253        &self,
254        ctx: &OperationContext,
255        from: &str,
256        to: &str,
257        args: OpCopy,
258    ) -> Result<Self::Copier> {
259        let core = self.core.clone();
260        let ctx = ctx.clone();
261        let from = from.to_string();
262        let to = to.to_string();
263        let source_content_length_hint = args.source_content_length_hint();
264
265        Ok(oio::OneShotCopier::new(async move {
266            let res = core.get_by_path(&ctx, &from).await?;
267            let file: AliyunDriveFile =
268                serde_json::from_reader(res.reader()).map_err(new_json_serialize_error)?;
269            let source_content_length =
270                file.size.or(source_content_length_hint).ok_or_else(|| {
271                    Error::new(
272                        ErrorKind::Unexpected,
273                        "Aliyun Drive source file does not contain a size",
274                    )
275                })?;
276
277            if from == to {
278                Ok(MetadataBuilder::file(source_content_length).build())
279            } else {
280                // copy can overwrite.
281                match core.get_by_path(&ctx, &to).await {
282                    Err(err) if err.kind() == ErrorKind::NotFound => {}
283                    Err(err) => Err(err)?,
284                    Ok(res) => {
285                        let file: AliyunDriveFile = serde_json::from_reader(res.reader())
286                            .map_err(new_json_serialize_error)?;
287                        core.delete_path(&ctx, &file.file_id).await?;
288                    }
289                }
290                // there is no direct copy in AliyunDrive.
291                // so we need to copy the path first and then rename it.
292                let parent_path = get_parent(&to);
293                let parent_file_id = core.ensure_dir_exists(&ctx, parent_path).await?;
294
295                // if from and to are going to be placed in the same folder,
296                // copy_path will fail as we cannot change the name during this action.
297                // it has to be auto renamed.
298                let auto_rename = file.parent_file_id == parent_file_id;
299                let res = core
300                    .copy_path(&ctx, &file.file_id, &parent_file_id, auto_rename)
301                    .await?;
302                let file: CopyResponse =
303                    serde_json::from_reader(res.reader()).map_err(new_json_serialize_error)?;
304                let file_id = file.file_id;
305
306                let from_name = get_basename(&from);
307                let to_name = get_basename(&to);
308
309                if from_name != to_name {
310                    core.update_path(&ctx, &file_id, to_name).await?;
311                }
312
313                Ok(MetadataBuilder::file(source_content_length).build())
314            }
315        }))
316    }
317
318    async fn stat(&self, ctx: &OperationContext, path: &str, _args: OpStat) -> Result<RpStat> {
319        let res = self.core.get_by_path(ctx, path).await?;
320        let file: AliyunDriveFile =
321            serde_json::from_reader(res.reader()).map_err(new_json_serialize_error)?;
322
323        if file.path_type == "folder" {
324            let mut meta = MetadataBuilder::dir();
325            meta.last_modified(file.updated_at.parse::<Timestamp>().map_err(|e| {
326                Error::new(ErrorKind::Unexpected, "parse last modified time").set_source(e)
327            })?);
328
329            return Ok(RpStat::new(meta.build()));
330        }
331
332        let size = file.size.ok_or_else(|| {
333            Error::new(
334                ErrorKind::Unexpected,
335                "aliyun drive stat response does not contain file size",
336            )
337        })?;
338        let mut meta = MetadataBuilder::file(size);
339        meta.last_modified(file.updated_at.parse::<Timestamp>().map_err(|e| {
340            Error::new(ErrorKind::Unexpected, "parse last modified time").set_source(e)
341        })?);
342        if let Some(v) = file.content_type {
343            meta.content_type(v);
344        }
345
346        Ok(RpStat::new(meta.build()))
347    }
348    fn read(&self, ctx: &OperationContext, path: &str, args: OpRead) -> Result<Self::Reader> {
349        let output: oio::StreamReader<AliyunDriveReader> = {
350            Ok(oio::StreamReader::new(AliyunDriveReader::new(
351                self.clone(),
352                ctx.clone(),
353                path,
354                args,
355            )))
356        }?;
357
358        Ok(output)
359    }
360
361    fn delete(&self, ctx: &OperationContext) -> Result<Self::Deleter> {
362        let output: oio::OneShotDeleter<AliyunDriveDeleter> = {
363            Ok(oio::OneShotDeleter::new(AliyunDriveDeleter::new(
364                self.core.clone(),
365                ctx.clone(),
366            )))
367        }?;
368
369        Ok(output)
370    }
371
372    fn list(&self, ctx: &OperationContext, path: &str, args: OpList) -> Result<Self::Lister> {
373        let output: oio::PageLister<AliyunDriveLister> = {
374            let l = AliyunDriveLister::new_with_path(
375                self.core.clone(),
376                ctx.clone(),
377                path.to_string(),
378                args.limit(),
379            );
380
381            Ok(oio::PageLister::new(l))
382        }?;
383
384        Ok(output)
385    }
386
387    fn write(&self, ctx: &OperationContext, path: &str, args: OpWrite) -> Result<Self::Writer> {
388        Ok(AliyunDriveLazyWriter::new(
389            self.core.clone(),
390            ctx.clone(),
391            path.to_string(),
392            args,
393        ))
394    }
395
396    async fn presign(
397        &self,
398        _ctx: &OperationContext,
399        _path: &str,
400        _args: OpPresign,
401    ) -> Result<RpPresign> {
402        Err(Error::new(
403            ErrorKind::Unsupported,
404            "operation is not supported",
405        ))
406    }
407}
opendal_service_aliyun_drive/backend.rs

opendal_service_aliyun_drive/
backend.rs
