opendal/services/dropbox/
core.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::default::Default;
19use std::fmt::Debug;
20use std::fmt::Formatter;
21use std::sync::Arc;
22
23use bytes::Buf;
24use bytes::Bytes;
25use chrono::DateTime;
26use chrono::Utc;
27use http::header;
28use http::header::CONTENT_LENGTH;
29use http::header::CONTENT_TYPE;
30use http::Request;
31use http::Response;
32use http::StatusCode;
33use serde::Deserialize;
34use serde::Serialize;
35use tokio::sync::Mutex;
36
37use super::error::parse_error;
38use crate::raw::*;
39use crate::*;
40
41pub struct DropboxCore {
42    pub info: Arc<AccessorInfo>,
43    pub root: String,
44    pub signer: Arc<Mutex<DropboxSigner>>,
45}
46
47impl Debug for DropboxCore {
48    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
49        f.debug_struct("DropboxCore")
50            .field("root", &self.root)
51            .finish()
52    }
53}
54
55impl DropboxCore {
56    fn build_path(&self, path: &str) -> String {
57        let path = build_rooted_abs_path(&self.root, path);
58        // For dropbox, even the path is a directory,
59        // we still need to remove the trailing slash.
60        path.trim_end_matches('/').to_string()
61    }
62
63    pub async fn sign<T>(&self, req: &mut Request<T>) -> Result<()> {
64        let mut signer = self.signer.lock().await;
65
66        // Access token is valid, use it directly.
67        if !signer.access_token.is_empty() && signer.expires_in > Utc::now() {
68            let value = format!("Bearer {}", signer.access_token)
69                .parse()
70                .expect("token must be valid header value");
71            req.headers_mut().insert(header::AUTHORIZATION, value);
72            return Ok(());
73        }
74
75        // Refresh invalid token.
76        let url = "https://api.dropboxapi.com/oauth2/token".to_string();
77
78        let content = format!(
79            "grant_type=refresh_token&refresh_token={}&client_id={}&client_secret={}",
80            signer.refresh_token, signer.client_id, signer.client_secret
81        );
82        let bs = Bytes::from(content);
83
84        let request = Request::post(&url)
85            .header(CONTENT_TYPE, "application/x-www-form-urlencoded")
86            .header(CONTENT_LENGTH, bs.len())
87            .body(Buffer::from(bs))
88            .map_err(new_request_build_error)?;
89
90        let resp = self.info.http_client().send(request).await?;
91        let body = resp.into_body();
92
93        let token: DropboxTokenResponse =
94            serde_json::from_reader(body.reader()).map_err(new_json_deserialize_error)?;
95
96        // Update signer after token refreshed.
97        signer.access_token.clone_from(&token.access_token);
98
99        // Refresh it 2 minutes earlier.
100        signer.expires_in = Utc::now()
101            + chrono::TimeDelta::try_seconds(token.expires_in as i64)
102                .expect("expires_in must be valid seconds")
103            - chrono::TimeDelta::try_seconds(120).expect("120 must be valid seconds");
104
105        let value = format!("Bearer {}", token.access_token)
106            .parse()
107            .expect("token must be valid header value");
108        req.headers_mut().insert(header::AUTHORIZATION, value);
109
110        Ok(())
111    }
112
113    pub async fn dropbox_get(
114        &self,
115        path: &str,
116        range: BytesRange,
117        _: &OpRead,
118    ) -> Result<Response<HttpBody>> {
119        let url: String = "https://content.dropboxapi.com/2/files/download".to_string();
120        let download_args = DropboxDownloadArgs {
121            path: build_rooted_abs_path(&self.root, path),
122        };
123        let request_payload =
124            serde_json::to_string(&download_args).map_err(new_json_serialize_error)?;
125
126        let mut req = Request::post(&url)
127            .header("Dropbox-API-Arg", request_payload)
128            .header(CONTENT_LENGTH, 0);
129
130        if !range.is_full() {
131            req = req.header(header::RANGE, range.to_header());
132        }
133
134        let mut request = req.body(Buffer::new()).map_err(new_request_build_error)?;
135
136        self.sign(&mut request).await?;
137        self.info.http_client().fetch(request).await
138    }
139
140    pub async fn dropbox_update(
141        &self,
142        path: &str,
143        size: Option<usize>,
144        args: &OpWrite,
145        body: Buffer,
146    ) -> Result<Response<Buffer>> {
147        let url = "https://content.dropboxapi.com/2/files/upload".to_string();
148        let dropbox_update_args = DropboxUploadArgs {
149            path: build_rooted_abs_path(&self.root, path),
150            ..Default::default()
151        };
152        let mut request_builder = Request::post(&url);
153        if let Some(size) = size {
154            request_builder = request_builder.header(CONTENT_LENGTH, size);
155        }
156        request_builder = request_builder.header(
157            CONTENT_TYPE,
158            args.content_type().unwrap_or("application/octet-stream"),
159        );
160
161        let mut request = request_builder
162            .header(
163                "Dropbox-API-Arg",
164                serde_json::to_string(&dropbox_update_args).map_err(new_json_serialize_error)?,
165            )
166            .body(body)
167            .map_err(new_request_build_error)?;
168
169        self.sign(&mut request).await?;
170        self.info.http_client().send(request).await
171    }
172
173    pub async fn dropbox_delete(&self, path: &str) -> Result<Response<Buffer>> {
174        let url = "https://api.dropboxapi.com/2/files/delete_v2".to_string();
175        let args = DropboxDeleteArgs {
176            path: self.build_path(path),
177        };
178
179        let bs = Bytes::from(serde_json::to_string(&args).map_err(new_json_serialize_error)?);
180
181        let mut request = Request::post(&url)
182            .header(CONTENT_TYPE, "application/json")
183            .header(CONTENT_LENGTH, bs.len())
184            .body(Buffer::from(bs))
185            .map_err(new_request_build_error)?;
186
187        self.sign(&mut request).await?;
188        self.info.http_client().send(request).await
189    }
190
191    pub async fn dropbox_create_folder(&self, path: &str) -> Result<RpCreateDir> {
192        let url = "https://api.dropboxapi.com/2/files/create_folder_v2".to_string();
193        let args = DropboxCreateFolderArgs {
194            path: self.build_path(path),
195        };
196
197        let bs = Bytes::from(serde_json::to_string(&args).map_err(new_json_serialize_error)?);
198
199        let mut request = Request::post(&url)
200            .header(CONTENT_TYPE, "application/json")
201            .header(CONTENT_LENGTH, bs.len())
202            .body(Buffer::from(bs))
203            .map_err(new_request_build_error)?;
204
205        self.sign(&mut request).await?;
206        let resp = self.info.http_client().send(request).await?;
207        let status = resp.status();
208        match status {
209            StatusCode::OK => Ok(RpCreateDir::default()),
210            _ => {
211                let err = parse_error(resp);
212                match err.kind() {
213                    ErrorKind::AlreadyExists => Ok(RpCreateDir::default()),
214                    _ => Err(err),
215                }
216            }
217        }
218    }
219
220    pub async fn dropbox_list(
221        &self,
222        path: &str,
223        recursive: bool,
224        limit: Option<usize>,
225    ) -> Result<Response<Buffer>> {
226        let url = "https://api.dropboxapi.com/2/files/list_folder".to_string();
227
228        // The default settings here align with the DropboxAPI default settings.
229        // Refer: https://www.dropbox.com/developers/documentation/http/documentation#files-list_folder
230        let args = DropboxListArgs {
231            path: self.build_path(path),
232            recursive,
233            limit: limit.unwrap_or(1000),
234        };
235
236        let bs = Bytes::from(serde_json::to_string(&args).map_err(new_json_serialize_error)?);
237
238        let mut request = Request::post(&url)
239            .header(CONTENT_TYPE, "application/json")
240            .header(CONTENT_LENGTH, bs.len())
241            .body(Buffer::from(bs))
242            .map_err(new_request_build_error)?;
243
244        self.sign(&mut request).await?;
245        self.info.http_client().send(request).await
246    }
247
248    pub async fn dropbox_list_continue(&self, cursor: &str) -> Result<Response<Buffer>> {
249        let url = "https://api.dropboxapi.com/2/files/list_folder/continue".to_string();
250
251        let args = DropboxListContinueArgs {
252            cursor: cursor.to_string(),
253        };
254
255        let bs = Bytes::from(serde_json::to_string(&args).map_err(new_json_serialize_error)?);
256
257        let mut request = Request::post(&url)
258            .header(CONTENT_TYPE, "application/json")
259            .header(CONTENT_LENGTH, bs.len())
260            .body(Buffer::from(bs))
261            .map_err(new_request_build_error)?;
262
263        self.sign(&mut request).await?;
264        self.info.http_client().send(request).await
265    }
266
267    pub async fn dropbox_copy(&self, from: &str, to: &str) -> Result<Response<Buffer>> {
268        let url = "https://api.dropboxapi.com/2/files/copy_v2".to_string();
269
270        let args = DropboxCopyArgs {
271            from_path: self.build_path(from),
272            to_path: self.build_path(to),
273        };
274
275        let bs = Bytes::from(serde_json::to_string(&args).map_err(new_json_serialize_error)?);
276
277        let mut request = Request::post(&url)
278            .header(CONTENT_TYPE, "application/json")
279            .header(CONTENT_LENGTH, bs.len())
280            .body(Buffer::from(bs))
281            .map_err(new_request_build_error)?;
282
283        self.sign(&mut request).await?;
284        self.info.http_client().send(request).await
285    }
286
287    pub async fn dropbox_move(&self, from: &str, to: &str) -> Result<Response<Buffer>> {
288        let url = "https://api.dropboxapi.com/2/files/move_v2".to_string();
289
290        let args = DropboxMoveArgs {
291            from_path: self.build_path(from),
292            to_path: self.build_path(to),
293        };
294
295        let bs = Bytes::from(serde_json::to_string(&args).map_err(new_json_serialize_error)?);
296
297        let mut request = Request::post(&url)
298            .header(CONTENT_TYPE, "application/json")
299            .header(CONTENT_LENGTH, bs.len())
300            .body(Buffer::from(bs))
301            .map_err(new_request_build_error)?;
302
303        self.sign(&mut request).await?;
304        self.info.http_client().send(request).await
305    }
306
307    pub async fn dropbox_get_metadata(&self, path: &str) -> Result<Response<Buffer>> {
308        let url = "https://api.dropboxapi.com/2/files/get_metadata".to_string();
309        let args = DropboxMetadataArgs {
310            path: self.build_path(path),
311            ..Default::default()
312        };
313
314        let bs = Bytes::from(serde_json::to_string(&args).map_err(new_json_serialize_error)?);
315
316        let mut request = Request::post(&url)
317            .header(CONTENT_TYPE, "application/json")
318            .header(CONTENT_LENGTH, bs.len())
319            .body(Buffer::from(bs))
320            .map_err(new_request_build_error)?;
321
322        self.sign(&mut request).await?;
323
324        self.info.http_client().send(request).await
325    }
326}
327
328#[derive(Clone)]
329pub struct DropboxSigner {
330    pub client_id: String,
331    pub client_secret: String,
332    pub refresh_token: String,
333
334    pub access_token: String,
335    pub expires_in: DateTime<Utc>,
336}
337
338impl Default for DropboxSigner {
339    fn default() -> Self {
340        DropboxSigner {
341            refresh_token: String::new(),
342            client_id: String::new(),
343            client_secret: String::new(),
344
345            access_token: String::new(),
346            expires_in: DateTime::<Utc>::MIN_UTC,
347        }
348    }
349}
350
351#[derive(Clone, Debug, Deserialize, Serialize)]
352struct DropboxDownloadArgs {
353    path: String,
354}
355
356#[derive(Clone, Debug, Deserialize, Serialize)]
357struct DropboxUploadArgs {
358    path: String,
359    mode: String,
360    mute: bool,
361    autorename: bool,
362    strict_conflict: bool,
363}
364
365impl Default for DropboxUploadArgs {
366    fn default() -> Self {
367        DropboxUploadArgs {
368            mode: "overwrite".to_string(),
369            path: "".to_string(),
370            mute: true,
371            autorename: false,
372            strict_conflict: false,
373        }
374    }
375}
376
377#[derive(Clone, Debug, Deserialize, Serialize)]
378struct DropboxDeleteArgs {
379    path: String,
380}
381
382#[derive(Clone, Debug, Deserialize, Serialize)]
383struct DropboxDeleteBatchEntry {
384    path: String,
385}
386
387#[derive(Clone, Debug, Deserialize, Serialize)]
388struct DropboxDeleteBatchArgs {
389    entries: Vec<DropboxDeleteBatchEntry>,
390}
391
392#[derive(Clone, Debug, Deserialize, Serialize)]
393struct DropboxDeleteBatchCheckArgs {
394    async_job_id: String,
395}
396
397#[derive(Clone, Debug, Deserialize, Serialize)]
398struct DropboxCreateFolderArgs {
399    path: String,
400}
401
402#[derive(Clone, Debug, Deserialize, Serialize)]
403struct DropboxListArgs {
404    path: String,
405    recursive: bool,
406    limit: usize,
407}
408
409#[derive(Clone, Debug, Deserialize, Serialize)]
410struct DropboxListContinueArgs {
411    cursor: String,
412}
413
414#[derive(Clone, Debug, Deserialize, Serialize)]
415struct DropboxCopyArgs {
416    from_path: String,
417    to_path: String,
418}
419
420#[derive(Clone, Debug, Deserialize, Serialize)]
421struct DropboxMoveArgs {
422    from_path: String,
423    to_path: String,
424}
425
426#[derive(Default, Clone, Debug, Deserialize, Serialize)]
427struct DropboxMetadataArgs {
428    include_deleted: bool,
429    include_has_explicit_shared_members: bool,
430    include_media_info: bool,
431    path: String,
432}
433
434#[derive(Clone, Deserialize)]
435struct DropboxTokenResponse {
436    access_token: String,
437    expires_in: usize,
438}
439
440#[derive(Default, Debug, Deserialize)]
441#[serde(default)]
442pub struct DropboxMetadataResponse {
443    #[serde(rename(deserialize = ".tag"))]
444    pub tag: String,
445    pub client_modified: String,
446    pub content_hash: Option<String>,
447    pub file_lock_info: Option<DropboxMetadataFileLockInfo>,
448    pub has_explicit_shared_members: Option<bool>,
449    pub id: String,
450    pub is_downloadable: Option<bool>,
451    pub name: String,
452    pub path_display: String,
453    pub path_lower: String,
454    pub property_groups: Option<Vec<DropboxMetadataPropertyGroup>>,
455    pub rev: Option<String>,
456    pub server_modified: Option<String>,
457    pub sharing_info: Option<DropboxMetadataSharingInfo>,
458    pub size: Option<u64>,
459}
460
461#[derive(Default, Debug, Deserialize)]
462#[serde(default)]
463pub struct DropboxMetadataFileLockInfo {
464    pub created: Option<String>,
465    pub is_lockholder: bool,
466    pub lockholder_name: Option<String>,
467}
468
469#[derive(Default, Debug, Deserialize)]
470#[serde(default)]
471pub struct DropboxMetadataPropertyGroup {
472    pub fields: Vec<DropboxMetadataPropertyGroupField>,
473    pub template_id: String,
474}
475
476#[derive(Default, Debug, Deserialize)]
477#[serde(default)]
478pub struct DropboxMetadataPropertyGroupField {
479    pub name: String,
480    pub value: String,
481}
482
483#[derive(Default, Debug, Deserialize)]
484#[serde(default)]
485pub struct DropboxMetadataSharingInfo {
486    pub modified_by: Option<String>,
487    pub parent_shared_folder_id: Option<String>,
488    pub read_only: Option<bool>,
489    pub shared_folder_id: Option<String>,
490    pub traverse_only: Option<bool>,
491    pub no_access: Option<bool>,
492}
493
494#[derive(Default, Debug, Deserialize)]
495#[serde(default)]
496pub struct DropboxListResponse {
497    pub entries: Vec<DropboxMetadataResponse>,
498    pub cursor: String,
499    pub has_more: bool,
500}
501
502#[derive(Default, Debug, Deserialize)]
503#[serde(default)]
504pub struct DropboxDeleteBatchResponse {
505    #[serde(rename(deserialize = ".tag"))]
506    pub tag: String,
507    pub async_job_id: Option<String>,
508    pub entries: Option<Vec<DropboxDeleteBatchResponseEntry>>,
509}
510
511#[derive(Default, Debug, Deserialize)]
512#[serde(default)]
513pub struct DropboxDeleteBatchResponseEntry {
514    #[serde(rename(deserialize = ".tag"))]
515    pub tag: String,
516    pub metadata: Option<DropboxMetadataResponse>,
517}
518
519#[derive(Default, Debug, Deserialize)]
520#[serde(default)]
521pub struct DropboxDeleteBatchFailureResponseCause {
522    #[serde(rename(deserialize = ".tag"))]
523    pub tag: String,
524}
opendal/services/dropbox/core.rs

opendal/services/dropbox/
core.rs
