opendal_service_gdrive/
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 http::StatusCode;
23
24use super::core::GdriveCore;
25use super::core::GdriveFile;
26use super::core::GdriveRecentPathState;
27use super::core::normalize_dir_path;
28use super::core::parse_error;
29use super::deleter::GdriveDeleter;
30use super::lister::GdriveFlatLister;
31use super::lister::GdriveLister;
32use super::reader::*;
33use super::writer::GdriveWriter;
34use opendal_core::raw::*;
35use opendal_core::*;
36
37use log::debug;
38use mea::mutex::Mutex;
39
40use super::GDRIVE_SCHEME;
41use super::config::GdriveConfig;
42use super::core::GdrivePathQuery;
43use super::core::GdriveSigner;
44use super::path_index::GdrivePathIndex;
45
46/// [GoogleDrive](https://drive.google.com/) backend support.
47#[derive(Default)]
48#[doc = include_str!("docs.md")]
49pub struct GdriveBuilder {
50    pub(super) config: GdriveConfig,
51}
52
53impl Debug for GdriveBuilder {
54    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
55        f.debug_struct("GdriveBuilder")
56            .field("config", &self.config)
57            .finish_non_exhaustive()
58    }
59}
60
61impl GdriveBuilder {
62    /// Set root path of GoogleDrive folder.
63    pub fn root(mut self, root: &str) -> Self {
64        self.config.root = if root.is_empty() {
65            None
66        } else {
67            Some(root.to_string())
68        };
69
70        self
71    }
72
73    /// Access token is used for temporary access to the GoogleDrive API.
74    ///
75    /// You can get the access token from [GoogleDrive App Console](https://console.cloud.google.com/apis/credentials)
76    /// or [GoogleDrive OAuth2 Playground](https://developers.google.com/oauthplayground/)
77    ///
78    /// # Note
79    ///
80    /// - An access token is valid for 1 hour.
81    /// - If you want to use the access token for a long time,
82    ///   you can use the refresh token to get a new access token.
83    pub fn access_token(mut self, access_token: &str) -> Self {
84        self.config.access_token = Some(access_token.to_string());
85        self
86    }
87
88    /// Refresh token is used for long term access to the GoogleDrive API.
89    ///
90    /// You can get the refresh token via OAuth 2.0 Flow of GoogleDrive API.
91    ///
92    /// OpenDAL will use this refresh token to get a new access token when the old one is expired.
93    pub fn refresh_token(mut self, refresh_token: &str) -> Self {
94        self.config.refresh_token = Some(refresh_token.to_string());
95        self
96    }
97
98    /// Set the client id for GoogleDrive.
99    ///
100    /// This is required for OAuth 2.0 Flow to refresh the access token.
101    pub fn client_id(mut self, client_id: &str) -> Self {
102        self.config.client_id = Some(client_id.to_string());
103        self
104    }
105
106    /// Set the client secret for GoogleDrive.
107    ///
108    /// This is required for OAuth 2.0 Flow with refresh the access token.
109    pub fn client_secret(mut self, client_secret: &str) -> Self {
110        self.config.client_secret = Some(client_secret.to_string());
111        self
112    }
113}
114
115impl Builder for GdriveBuilder {
116    type Config = GdriveConfig;
117
118    fn build(self) -> Result<impl Service> {
119        let root = normalize_root(&self.config.root.unwrap_or_default());
120        debug!("backend use root {root}");
121
122        let info = ServiceInfo::new(GDRIVE_SCHEME, &root, "");
123        let capability = Capability {
124            stat: true,
125
126            read: true,
127            read_with_suffix: true,
128
129            list: true,
130            list_with_recursive: true,
131
132            write: true,
133
134            create_dir: true,
135            delete: true,
136            delete_with_recursive: true,
137            rename: true,
138            copy: true,
139
140            shared: true,
141
142            ..Default::default()
143        };
144
145        let accessor_info = info;
146        let mut signer = GdriveSigner::new();
147        match (self.config.access_token, self.config.refresh_token) {
148            (Some(access_token), None) => {
149                signer.access_token = access_token;
150                // We will never expire user specified access token.
151                signer.expires_in = Timestamp::MAX;
152            }
153            (None, Some(refresh_token)) => {
154                let client_id = self.config.client_id.ok_or_else(|| {
155                    Error::new(
156                        ErrorKind::ConfigInvalid,
157                        "client_id must be set when refresh_token is set",
158                    )
159                    .with_context("service", GDRIVE_SCHEME)
160                })?;
161                let client_secret = self.config.client_secret.ok_or_else(|| {
162                    Error::new(
163                        ErrorKind::ConfigInvalid,
164                        "client_secret must be set when refresh_token is set",
165                    )
166                    .with_context("service", GDRIVE_SCHEME)
167                })?;
168
169                signer.refresh_token = refresh_token;
170                signer.client_id = client_id;
171                signer.client_secret = client_secret;
172            }
173            (Some(_), Some(_)) => {
174                return Err(Error::new(
175                    ErrorKind::ConfigInvalid,
176                    "access_token and refresh_token cannot be set at the same time",
177                )
178                .with_context("service", GDRIVE_SCHEME));
179            }
180            (None, None) => {
181                return Err(Error::new(
182                    ErrorKind::ConfigInvalid,
183                    "access_token or refresh_token must be set",
184                )
185                .with_context("service", GDRIVE_SCHEME));
186            }
187        };
188
189        let signer = Arc::new(Mutex::new(signer));
190
191        Ok(GdriveBackend {
192            core: Arc::new(GdriveCore {
193                info: accessor_info.clone(),
194                capability,
195                root,
196                signer: signer.clone(),
197                path_index: GdrivePathIndex::new(GdrivePathQuery::new(signer)),
198                recent_entries: Mutex::default(),
199            }),
200        })
201    }
202}
203
204#[derive(Clone, Debug)]
205pub struct GdriveBackend {
206    pub core: Arc<GdriveCore>,
207}
208
209/// Lister type that supports both recursive and non-recursive listing
210pub type GdriveListers = TwoWays<oio::PageLister<GdriveLister>, GdriveFlatLister>;
211
212impl Service for GdriveBackend {
213    type Reader = oio::StreamReader<GdriveReader>;
214    type Writer = oio::OneShotWriter<GdriveWriter>;
215    type Lister = GdriveListers;
216    type Deleter = oio::OneShotDeleter<GdriveDeleter>;
217    type Copier = oio::OneShotCopier;
218
219    fn info(&self) -> ServiceInfo {
220        self.core.info.clone()
221    }
222
223    fn capability(&self) -> Capability {
224        self.core.capability
225    }
226
227    async fn create_dir(
228        &self,
229        ctx: &OperationContext,
230        path: &str,
231        _args: OpCreateDir,
232    ) -> Result<RpCreateDir> {
233        let path = build_abs_path(&self.core.root, path);
234        let dir_id = self.core.ensure_dir(ctx, &path).await?;
235        let metadata = Metadata::new(EntryMode::DIR);
236
237        self.core.cache_dir_id(&path, &dir_id).await;
238        self.core.record_recent_upsert(&path, metadata).await;
239
240        Ok(RpCreateDir::default())
241    }
242
243    async fn stat(&self, ctx: &OperationContext, path: &str, _args: OpStat) -> Result<RpStat> {
244        let path = build_abs_path(&self.core.root, path);
245
246        match self.core.recent_entry_for_path(&path).await {
247            GdriveRecentPathState::Present(metadata) => {
248                if metadata.mode().is_dir() && path.ends_with('/') {
249                    return Ok(RpStat::new(*metadata));
250                }
251            }
252            GdriveRecentPathState::Deleted => {
253                return Err(Error::new(
254                    ErrorKind::NotFound,
255                    format!("path not found: {path}"),
256                ));
257            }
258            GdriveRecentPathState::Missing => {}
259        }
260
261        let mut file_id = match self.core.resolve_path(ctx, &path).await? {
262            Some(id) => id,
263            None => match self.core.resolve_path_after_refresh(ctx, &path).await? {
264                Some(id) => id,
265                None => {
266                    return Err(Error::new(
267                        ErrorKind::NotFound,
268                        format!("path not found: {path}"),
269                    ));
270                }
271            },
272        };
273        let mut resp = self.core.gdrive_stat_by_id(ctx, &file_id).await?;
274
275        if resp.status() == StatusCode::NOT_FOUND {
276            file_id = self
277                .core
278                .resolve_path_after_refresh(ctx, &path)
279                .await?
280                .ok_or(Error::new(
281                    ErrorKind::NotFound,
282                    format!("path not found: {path}"),
283                ))?;
284            resp = self.core.gdrive_stat_by_id(ctx, &file_id).await?;
285        }
286
287        if resp.status() != StatusCode::OK {
288            return Err(parse_error(resp));
289        }
290
291        let bs = resp.into_body();
292        let gdrive_file: GdriveFile =
293            serde_json::from_reader(bs.reader()).map_err(new_json_deserialize_error)?;
294
295        let file_type = if gdrive_file.mime_type == "application/vnd.google-apps.folder" {
296            EntryMode::DIR
297        } else {
298            EntryMode::FILE
299        };
300        let mut meta = Metadata::new(file_type).with_content_type(gdrive_file.mime_type);
301        if let Some(v) = gdrive_file.size {
302            meta = meta.with_content_length(v.parse::<u64>().map_err(|e| {
303                Error::new(ErrorKind::Unexpected, "parse content length").set_source(e)
304            })?);
305        }
306        if let Some(v) = gdrive_file.modified_time {
307            meta = meta.with_last_modified(v.parse::<Timestamp>().map_err(|e| {
308                Error::new(ErrorKind::Unexpected, "parse last modified time").set_source(e)
309            })?);
310        }
311        Ok(RpStat::new(meta))
312    }
313    fn read(&self, ctx: &OperationContext, path: &str, args: OpRead) -> Result<Self::Reader> {
314        let output: oio::StreamReader<GdriveReader> = {
315            Ok(oio::StreamReader::new(GdriveReader::new(
316                self.clone(),
317                ctx.clone(),
318                path,
319                args,
320            )))
321        }?;
322
323        Ok(output)
324    }
325
326    fn write(&self, ctx: &OperationContext, path: &str, _: OpWrite) -> Result<Self::Writer> {
327        let output: oio::OneShotWriter<GdriveWriter> = {
328            let path = build_abs_path(&self.core.root, path);
329
330            Ok(oio::OneShotWriter::new(GdriveWriter::new(
331                self.core.clone(),
332                ctx.clone(),
333                path,
334                None,
335            )))
336        }?;
337
338        Ok(output)
339    }
340
341    fn delete(&self, ctx: &OperationContext) -> Result<Self::Deleter> {
342        let output: oio::OneShotDeleter<GdriveDeleter> = {
343            Ok(oio::OneShotDeleter::new(GdriveDeleter::new(
344                self.core.clone(),
345                ctx.clone(),
346            )))
347        }?;
348
349        Ok(output)
350    }
351
352    fn list(&self, ctx: &OperationContext, path: &str, args: OpList) -> Result<Self::Lister> {
353        let output: GdriveListers = {
354            let path = build_abs_path(&self.core.root, path);
355
356            if args.recursive() {
357                // Use optimized batch-query recursive lister
358                let l = GdriveFlatLister::new(path, self.core.clone(), ctx.clone());
359                Ok(TwoWays::Two(l))
360            } else {
361                // Use standard page-based lister for non-recursive
362                let l = GdriveLister::new(path, self.core.clone(), ctx.clone());
363                Ok(TwoWays::One(oio::PageLister::new(l)))
364            }
365        }?;
366
367        Ok(output)
368    }
369
370    fn copy(
371        &self,
372        ctx: &OperationContext,
373        from: &str,
374        to: &str,
375        _args: OpCopy,
376        _opts: OpCopier,
377    ) -> Result<Self::Copier> {
378        let core = self.core.clone();
379        let ctx = ctx.clone();
380        let from = from.to_string();
381        let to = to.to_string();
382
383        Ok(oio::OneShotCopier::new(async move {
384            let source = build_abs_path(&core.root, &from);
385            let target = build_abs_path(&core.root, &to);
386            let resp = core.gdrive_copy(&ctx, &from, &to).await?;
387
388            match resp.status() {
389                StatusCode::OK => {
390                    let body = resp.into_body();
391                    let meta: GdriveFile = serde_json::from_reader(body.reader())
392                        .map_err(new_json_deserialize_error)?;
393
394                    let to_path = build_abs_path(&core.root, &to);
395                    let mut metadata = if meta.mime_type == "application/vnd.google-apps.folder" {
396                        Metadata::new(EntryMode::DIR)
397                    } else {
398                        Metadata::new(EntryMode::FILE)
399                    };
400                    if let Some(size) = meta.size {
401                        metadata =
402                            metadata.with_content_length(size.parse::<u64>().map_err(|e| {
403                                Error::new(ErrorKind::Unexpected, "parse content length")
404                                    .set_source(e)
405                            })?);
406                    }
407
408                    if metadata.mode().is_dir() {
409                        core.cache_dir_id(&to_path, &meta.id).await;
410                    } else {
411                        core.cache_file_id(&to_path, &meta.id).await;
412                    }
413                    core.record_recent_upsert(&to_path, metadata).await;
414
415                    Ok(Metadata::default())
416                }
417                StatusCode::NOT_FOUND => {
418                    core.refresh_path(&source).await;
419                    core.refresh_path(&target).await;
420                    let resp = core.gdrive_copy(&ctx, &from, &to).await?;
421                    match resp.status() {
422                        StatusCode::OK => {
423                            let body = resp.into_body();
424                            let meta: GdriveFile = serde_json::from_reader(body.reader())
425                                .map_err(new_json_deserialize_error)?;
426
427                            let to_path = build_abs_path(&core.root, &to);
428                            let mut metadata =
429                                if meta.mime_type == "application/vnd.google-apps.folder" {
430                                    Metadata::new(EntryMode::DIR)
431                                } else {
432                                    Metadata::new(EntryMode::FILE)
433                                };
434                            if let Some(size) = meta.size {
435                                metadata = metadata.with_content_length(
436                                    size.parse::<u64>().map_err(|e| {
437                                        Error::new(ErrorKind::Unexpected, "parse content length")
438                                            .set_source(e)
439                                    })?,
440                                );
441                            }
442
443                            if metadata.mode().is_dir() {
444                                core.cache_dir_id(&to_path, &meta.id).await;
445                            } else {
446                                core.cache_file_id(&to_path, &meta.id).await;
447                            }
448                            core.record_recent_upsert(&to_path, metadata).await;
449
450                            Ok(Metadata::default())
451                        }
452                        _ => Err(parse_error(resp)),
453                    }
454                }
455                _ => Err(parse_error(resp)),
456            }
457        }))
458    }
459
460    async fn rename(
461        &self,
462        ctx: &OperationContext,
463        from: &str,
464        to: &str,
465        _args: OpRename,
466    ) -> Result<RpRename> {
467        let source = build_abs_path(&self.core.root, from);
468        let target = build_abs_path(&self.core.root, to);
469
470        // rename will overwrite `to`, delete it if exist
471        self.core.trash_path_if_exists(ctx, &target).await?;
472
473        let resp = self
474            .core
475            .gdrive_patch_metadata_request(ctx, &source, &target)
476            .await?;
477
478        let status = resp.status();
479
480        match status {
481            StatusCode::OK => {
482                let body = resp.into_body();
483                let meta: GdriveFile =
484                    serde_json::from_reader(body.reader()).map_err(new_json_deserialize_error)?;
485
486                let source_path = if meta.mime_type == "application/vnd.google-apps.folder" {
487                    normalize_dir_path(&build_abs_path(&self.core.root, from))
488                } else {
489                    build_abs_path(&self.core.root, from)
490                };
491                let target_path = if meta.mime_type == "application/vnd.google-apps.folder" {
492                    normalize_dir_path(&build_abs_path(&self.core.root, to))
493                } else {
494                    build_abs_path(&self.core.root, to)
495                };
496                let mut metadata = if meta.mime_type == "application/vnd.google-apps.folder" {
497                    Metadata::new(EntryMode::DIR)
498                } else {
499                    Metadata::new(EntryMode::FILE)
500                };
501                if let Some(size) = meta.size {
502                    metadata = metadata.with_content_length(size.parse::<u64>().map_err(|e| {
503                        Error::new(ErrorKind::Unexpected, "parse content length").set_source(e)
504                    })?);
505                }
506
507                if metadata.mode().is_dir() {
508                    self.core.invalidate_dir_id(&source_path).await;
509                    self.core.cache_dir_id(&target_path, &meta.id).await;
510                } else {
511                    self.core.invalidate_file_id(&source_path).await;
512                    self.core.cache_file_id(&target_path, &meta.id).await;
513                }
514                self.core
515                    .record_recent_delete(&source_path, metadata.mode())
516                    .await;
517                self.core.record_recent_upsert(&target_path, metadata).await;
518
519                Ok(RpRename::default())
520            }
521            StatusCode::NOT_FOUND => {
522                self.core.refresh_path(&source).await;
523                self.core.refresh_path(&target).await;
524
525                let resp = self
526                    .core
527                    .gdrive_patch_metadata_request(ctx, &source, &target)
528                    .await?;
529
530                if resp.status() != StatusCode::OK {
531                    return Err(parse_error(resp));
532                }
533
534                let body = resp.into_body();
535                let meta: GdriveFile =
536                    serde_json::from_reader(body.reader()).map_err(new_json_deserialize_error)?;
537
538                let source_path = if meta.mime_type == "application/vnd.google-apps.folder" {
539                    normalize_dir_path(&build_abs_path(&self.core.root, from))
540                } else {
541                    build_abs_path(&self.core.root, from)
542                };
543                let target_path = if meta.mime_type == "application/vnd.google-apps.folder" {
544                    normalize_dir_path(&build_abs_path(&self.core.root, to))
545                } else {
546                    build_abs_path(&self.core.root, to)
547                };
548                let mut metadata = if meta.mime_type == "application/vnd.google-apps.folder" {
549                    Metadata::new(EntryMode::DIR)
550                } else {
551                    Metadata::new(EntryMode::FILE)
552                };
553                if let Some(size) = meta.size {
554                    metadata = metadata.with_content_length(size.parse::<u64>().map_err(|e| {
555                        Error::new(ErrorKind::Unexpected, "parse content length").set_source(e)
556                    })?);
557                }
558
559                if metadata.mode().is_dir() {
560                    self.core.invalidate_dir_id(&source_path).await;
561                    self.core.cache_dir_id(&target_path, &meta.id).await;
562                } else {
563                    self.core.invalidate_file_id(&source_path).await;
564                    self.core.cache_file_id(&target_path, &meta.id).await;
565                }
566                self.core
567                    .record_recent_delete(&source_path, metadata.mode())
568                    .await;
569                self.core.record_recent_upsert(&target_path, metadata).await;
570
571                Ok(RpRename::default())
572            }
573            _ => Err(parse_error(resp)),
574        }
575    }
576
577    async fn presign(
578        &self,
579        _ctx: &OperationContext,
580        _path: &str,
581        _args: OpPresign,
582    ) -> Result<RpPresign> {
583        Err(Error::new(
584            ErrorKind::Unsupported,
585            "operation is not supported",
586        ))
587    }
588}
opendal_service_gdrive/backend.rs

opendal_service_gdrive/
backend.rs
