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 bytes::Buf;
22use log::debug;
23use mea::mutex::Mutex;
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
193    fn info(&self) -> ServiceInfo {
194        self.core.info.clone()
195    }
196
197    fn capability(&self) -> Capability {
198        self.core.capability
199    }
200
201    async fn create_dir(
202        &self,
203        ctx: &OperationContext,
204        path: &str,
205        _args: OpCreateDir,
206    ) -> Result<RpCreateDir> {
207        self.core.ensure_dir_exists(ctx, path).await?;
208
209        Ok(RpCreateDir::default())
210    }
211
212    async fn rename(
213        &self,
214        ctx: &OperationContext,
215        from: &str,
216        to: &str,
217        _args: OpRename,
218    ) -> Result<RpRename> {
219        if from == to {
220            return Ok(RpRename::default());
221        }
222        let res = self.core.get_by_path(ctx, from).await?;
223        let file: AliyunDriveFile =
224            serde_json::from_reader(res.reader()).map_err(new_json_serialize_error)?;
225        // rename can overwrite.
226        match self.core.get_by_path(ctx, to).await {
227            Err(err) if err.kind() == ErrorKind::NotFound => {}
228            Err(err) => return Err(err),
229            Ok(res) => {
230                let file: AliyunDriveFile =
231                    serde_json::from_reader(res.reader()).map_err(new_json_serialize_error)?;
232                self.core.delete_path(ctx, &file.file_id).await?;
233            }
234        };
235
236        let parent_file_id = self.core.ensure_dir_exists(ctx, get_parent(to)).await?;
237        self.core
238            .move_path(ctx, &file.file_id, &parent_file_id)
239            .await?;
240
241        let from_name = get_basename(from);
242        let to_name = get_basename(to);
243
244        if from_name != to_name {
245            self.core.update_path(ctx, &file.file_id, to_name).await?;
246        }
247
248        Ok(RpRename::default())
249    }
250
251    fn copy(
252        &self,
253        ctx: &OperationContext,
254        from: &str,
255        to: &str,
256        _args: OpCopy,
257        _opts: OpCopier,
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
264        Ok(oio::OneShotCopier::new(async move {
265            if from == to {
266                Ok(Metadata::default())
267            } else {
268                let res = core.get_by_path(&ctx, &from).await?;
269                let file: AliyunDriveFile =
270                    serde_json::from_reader(res.reader()).map_err(new_json_serialize_error)?;
271                // copy can overwrite.
272                match core.get_by_path(&ctx, &to).await {
273                    Err(err) if err.kind() == ErrorKind::NotFound => {}
274                    Err(err) => Err(err)?,
275                    Ok(res) => {
276                        let file: AliyunDriveFile = serde_json::from_reader(res.reader())
277                            .map_err(new_json_serialize_error)?;
278                        core.delete_path(&ctx, &file.file_id).await?;
279                    }
280                }
281                // there is no direct copy in AliyunDrive.
282                // so we need to copy the path first and then rename it.
283                let parent_path = get_parent(&to);
284                let parent_file_id = core.ensure_dir_exists(&ctx, parent_path).await?;
285
286                // if from and to are going to be placed in the same folder,
287                // copy_path will fail as we cannot change the name during this action.
288                // it has to be auto renamed.
289                let auto_rename = file.parent_file_id == parent_file_id;
290                let res = core
291                    .copy_path(&ctx, &file.file_id, &parent_file_id, auto_rename)
292                    .await?;
293                let file: CopyResponse =
294                    serde_json::from_reader(res.reader()).map_err(new_json_serialize_error)?;
295                let file_id = file.file_id;
296
297                let from_name = get_basename(&from);
298                let to_name = get_basename(&to);
299
300                if from_name != to_name {
301                    core.update_path(&ctx, &file_id, to_name).await?;
302                }
303
304                Ok(Metadata::default())
305            }
306        }))
307    }
308
309    async fn stat(&self, ctx: &OperationContext, path: &str, _args: OpStat) -> Result<RpStat> {
310        let res = self.core.get_by_path(ctx, path).await?;
311        let file: AliyunDriveFile =
312            serde_json::from_reader(res.reader()).map_err(new_json_serialize_error)?;
313
314        if file.path_type == "folder" {
315            let meta = Metadata::new(EntryMode::DIR).with_last_modified(
316                file.updated_at.parse::<Timestamp>().map_err(|e| {
317                    Error::new(ErrorKind::Unexpected, "parse last modified time").set_source(e)
318                })?,
319            );
320
321            return Ok(RpStat::new(meta));
322        }
323
324        let mut meta = Metadata::new(EntryMode::FILE).with_last_modified(
325            file.updated_at.parse::<Timestamp>().map_err(|e| {
326                Error::new(ErrorKind::Unexpected, "parse last modified time").set_source(e)
327            })?,
328        );
329        if let Some(v) = file.size {
330            meta = meta.with_content_length(v);
331        }
332        if let Some(v) = file.content_type {
333            meta = meta.with_content_type(v);
334        }
335
336        Ok(RpStat::new(meta))
337    }
338    fn read(&self, ctx: &OperationContext, path: &str, args: OpRead) -> Result<Self::Reader> {
339        let output: oio::StreamReader<AliyunDriveReader> = {
340            Ok(oio::StreamReader::new(AliyunDriveReader::new(
341                self.clone(),
342                ctx.clone(),
343                path,
344                args,
345            )))
346        }?;
347
348        Ok(output)
349    }
350
351    fn delete(&self, ctx: &OperationContext) -> Result<Self::Deleter> {
352        let output: oio::OneShotDeleter<AliyunDriveDeleter> = {
353            Ok(oio::OneShotDeleter::new(AliyunDriveDeleter::new(
354                self.core.clone(),
355                ctx.clone(),
356            )))
357        }?;
358
359        Ok(output)
360    }
361
362    fn list(&self, ctx: &OperationContext, path: &str, args: OpList) -> Result<Self::Lister> {
363        let output: oio::PageLister<AliyunDriveLister> = {
364            let l = AliyunDriveLister::new_with_path(
365                self.core.clone(),
366                ctx.clone(),
367                path.to_string(),
368                args.limit(),
369            );
370
371            Ok(oio::PageLister::new(l))
372        }?;
373
374        Ok(output)
375    }
376
377    fn write(&self, ctx: &OperationContext, path: &str, args: OpWrite) -> Result<Self::Writer> {
378        Ok(AliyunDriveLazyWriter::new(
379            self.core.clone(),
380            ctx.clone(),
381            path.to_string(),
382            args,
383        ))
384    }
385
386    async fn presign(
387        &self,
388        _ctx: &OperationContext,
389        _path: &str,
390        _args: OpPresign,
391    ) -> Result<RpPresign> {
392        Err(Error::new(
393            ErrorKind::Unsupported,
394            "operation is not supported",
395        ))
396    }
397}
opendal_service_aliyun_drive/backend.rs

opendal_service_aliyun_drive/
backend.rs
