feat(services/dropbox): impl create_dir and polish error handling (ap…

…ache#2600) * test(services/dropbox): enable CI test for dropbox Signed-off-by: suyanhanx <[email protected]> * try Signed-off-by: suyanhanx <[email protected]> * try Signed-off-by: suyanhanx <[email protected]> * fix Signed-off-by: suyanhanx <[email protected]> * support create_dir Signed-off-by: suyanhanx <[email protected]> * fix root not affected Signed-off-by: suyanhanx <[email protected]> * remove nextest Signed-off-by: suyanhanx <[email protected]> * fix Signed-off-by: suyanhanx <[email protected]> * fix Signed-off-by: suyanhanx <[email protected]> * tip for test run Signed-off-by: suyanhanx <[email protected]> * make fmt happy Signed-off-by: suyanhanx <[email protected]> * make clippy happy Signed-off-by: suyanhanx <[email protected]> * remove CI config file Signed-off-by: suyanhanx <[email protected]> * polish Signed-off-by: suyanhanx <[email protected]> --------- Signed-off-by: suyanhanx <[email protected]>
splitgraph · Jul 6, 2023 · a2634c4 · a2634c4
1 parent ed90281
commit a2634c4
Show file tree

Hide file tree

Showing 5 changed files with 117 additions and 14 deletions.
diff --git a/.env.example b/.env.example
@@ -123,3 +123,7 @@ OPENDAL_REDB_TABLE=redb-table
 # cacache
 OPENDAL_CACACHE_TEST=false
 OPENDAL_CACACHE_DATADIR=/tmp/opendal/cacache/
+#dropbox
+OPENDAL_DROPBOX_TEST=false
+OPENDAL_DROPBOX_ROOT=/tmp/opendal/
+OPENDAL_DROPBOX_ACCESS_TOKEN=<access_token>
diff --git a/core/src/services/dropbox/backend.rs b/core/src/services/dropbox/backend.rs
@@ -48,14 +48,36 @@ impl Accessor for DropboxBackend {
         ma.set_scheme(Scheme::Dropbox)
             .set_root(&self.core.root)
             .set_capability(Capability {
+                stat: true,
+
                 read: true,
+
                 write: true,
+
+                create_dir: true,
+
                 delete: true,
+
                 ..Default::default()
             });
         ma
     }
 
+    async fn create_dir(&self, path: &str, _args: OpCreateDir) -> Result<RpCreateDir> {
+        let resp = self.core.dropbox_create_folder(path).await?;
+        let status = resp.status();
+        match status {
+            StatusCode::OK => Ok(RpCreateDir::default()),
+            _ => {
+                let err = parse_error(resp).await?;
+                match err.kind() {
+                    ErrorKind::AlreadyExists => Ok(RpCreateDir::default()),
+                    _ => Err(err),
+                }
+            }
+        }
+    }
+
     async fn read(&self, path: &str, _args: OpRead) -> Result<(RpRead, Self::Reader)> {
         let resp = self.core.dropbox_get(path).await?;
         let status = resp.status();
@@ -88,7 +110,13 @@ impl Accessor for DropboxBackend {
 
         match status {
             StatusCode::OK => Ok(RpDelete::default()),
-            _ => Err(parse_error(resp).await?),
+            _ => {
+                let err = parse_error(resp).await?;
+                match err.kind() {
+                    ErrorKind::NotFound => Ok(RpDelete::default()),
+                    _ => Err(err),
+                }
+            }
         }
     }
 
@@ -109,12 +137,22 @@ impl Accessor for DropboxBackend {
                     _ => EntryMode::Unknown,
                 };
                 let mut metadata = Metadata::new(entry_mode);
-                let last_modified = decoded_response.client_modified;
-                let date_utc_last_modified = parse_datetime_from_rfc3339(&last_modified)?;
-                metadata.set_last_modified(date_utc_last_modified);
-                if decoded_response.size.is_some() {
-                    let size = decoded_response.size.unwrap();
-                    metadata.set_content_length(size);
+                // Only set last_modified and size if entry_mode is FILE, because Dropbox API
+                // returns last_modified and size only for files.
+                // FYI: https://www.dropbox.com/developers/documentation/http/documentation#files-get_metadata
+                if entry_mode == EntryMode::FILE {
+                    let date_utc_last_modified =
+                        parse_datetime_from_rfc3339(&decoded_response.client_modified)?;
+                    metadata.set_last_modified(date_utc_last_modified);
+
+                    if let Some(size) = decoded_response.size {
+                        metadata.set_content_length(size);
+                    } else {
+                        return Err(Error::new(
+                            ErrorKind::Unexpected,
+                            &format!("no size found for file {}", path),
+                        ));
+                    }
                 }
                 Ok(RpStat::new(metadata))
             }

diff --git a/core/src/services/dropbox/builder.rs b/core/src/services/dropbox/builder.rs
@@ -114,6 +114,7 @@ impl Builder for DropboxBuilder {
 
     fn from_map(map: HashMap<String, String>) -> Self {
         let mut builder = Self::default();
+        map.get("root").map(|v| builder.root(v));
         map.get("access_token").map(|v| builder.access_token(v));
         builder
     }

diff --git a/core/src/services/dropbox/core.rs b/core/src/services/dropbox/core.rs
@@ -49,6 +49,13 @@ impl Debug for DropboxCore {
 }
 
 impl DropboxCore {
+    fn build_path(&self, path: &str) -> String {
+        let path = build_rooted_abs_path(&self.root, path);
+        // For dropbox, even the path is a directory,
+        // we still need to remove the trailing slash.
+        path.trim_end_matches('/').to_string()
+    }
+
     pub async fn dropbox_get(&self, path: &str) -> Result<Response<IncomingAsyncBody>> {
         let url: String = "https://content.dropboxapi.com/2/files/download".to_string();
         let download_args = DropboxDownloadArgs {
@@ -59,6 +66,7 @@ impl DropboxCore {
         let request = self
             .build_auth_header(Request::post(&url))
             .header("Dropbox-API-Arg", request_payload)
+            .header(header::CONTENT_LENGTH, 0)
             .body(AsyncBody::Empty)
             .map_err(new_request_build_error)?;
         self.client.send(request).await
@@ -80,9 +88,11 @@ impl DropboxCore {
         if let Some(size) = size {
             request_builder = request_builder.header(header::CONTENT_LENGTH, size);
         }
-        if let Some(mime) = content_type {
-            request_builder = request_builder.header(header::CONTENT_TYPE, mime);
-        }
+        request_builder = request_builder.header(
+            header::CONTENT_TYPE,
+            content_type.unwrap_or("application/octet-stream"),
+        );
+
         let request = self
             .build_auth_header(request_builder)
             .header(
@@ -98,14 +108,32 @@ impl DropboxCore {
     pub async fn dropbox_delete(&self, path: &str) -> Result<Response<IncomingAsyncBody>> {
         let url = "https://api.dropboxapi.com/2/files/delete_v2".to_string();
         let args = DropboxDeleteArgs {
-            path: build_rooted_abs_path(&self.root, path),
+            path: self.build_path(path),
+        };
+
+        let bs = Bytes::from(serde_json::to_string(&args).map_err(new_json_serialize_error)?);
+
+        let request = self
+            .build_auth_header(Request::post(&url))
+            .header(header::CONTENT_TYPE, "application/json")
+            .header(header::CONTENT_LENGTH, bs.len())
+            .body(AsyncBody::Bytes(bs))
+            .map_err(new_request_build_error)?;
+        self.client.send(request).await
+    }
+
+    pub async fn dropbox_create_folder(&self, path: &str) -> Result<Response<IncomingAsyncBody>> {
+        let url = "https://api.dropboxapi.com/2/files/create_folder_v2".to_string();
+        let args = DropboxCreateFolderArgs {
+            path: self.build_path(path),
         };
 
         let bs = Bytes::from(serde_json::to_string(&args).map_err(new_json_serialize_error)?);
 
         let request = self
             .build_auth_header(Request::post(&url))
             .header(header::CONTENT_TYPE, "application/json")
+            .header(header::CONTENT_LENGTH, bs.len())
             .body(AsyncBody::Bytes(bs))
             .map_err(new_request_build_error)?;
         self.client.send(request).await
@@ -114,7 +142,7 @@ impl DropboxCore {
     pub async fn dropbox_get_metadata(&self, path: &str) -> Result<Response<IncomingAsyncBody>> {
         let url = "https://api.dropboxapi.com/2/files/get_metadata".to_string();
         let args = DropboxMetadataArgs {
-            path: build_rooted_abs_path(&self.root, path),
+            path: self.build_path(path),
             ..Default::default()
         };
 
@@ -123,6 +151,7 @@ impl DropboxCore {
         let request = self
             .build_auth_header(Request::post(&url))
             .header(header::CONTENT_TYPE, "application/json")
+            .header(header::CONTENT_LENGTH, bs.len())
             .body(AsyncBody::Bytes(bs))
             .map_err(new_request_build_error)?;
         self.client.send(request).await
@@ -154,6 +183,11 @@ struct DropboxDeleteArgs {
     path: String,
 }
 
+#[derive(Clone, Debug, Deserialize, Serialize)]
+struct DropboxCreateFolderArgs {
+    path: String,
+}
+
 #[derive(Clone, Debug, Deserialize, Serialize)]
 struct DropboxMetadataArgs {
     include_deleted: bool,

diff --git a/core/src/services/dropbox/error.rs b/core/src/services/dropbox/error.rs
@@ -38,12 +38,38 @@ pub async fn parse_error(resp: Response<IncomingAsyncBody>) -> Result<Error> {
         | StatusCode::GATEWAY_TIMEOUT => (ErrorKind::Unexpected, true),
         _ => (ErrorKind::Unexpected, false),
     };
+
     let dropbox_error =
         serde_json::from_slice::<DropboxErrorResponse>(&bs).map_err(new_json_deserialize_error);
     match dropbox_error {
         Ok(dropbox_error) => {
-            let mut err = Error::new(kind, dropbox_error.error_summary.as_ref())
-                .with_context("response", format!("{parts:?}"));
+            // We cannot get the error type from the response header when the status code is 409.
+            // Because Dropbox API v2 will put error summary in the response body,
+            // we need to parse it to get the correct error type and then error kind.
+            // See https://www.dropbox.com/developers/documentation/http/documentation#error-handling
+            let error_summary = dropbox_error.error_summary.as_str();
+
+            let mut err = Error::new(
+                match parts.status {
+                    // 409 Conflict means that Endpoint-specific error.
+                    // Look to the JSON response body for the specifics of the error.
+                    StatusCode::CONFLICT => {
+                        if error_summary.contains("path/not_found")
+                            || error_summary.contains("path_lookup/not_found")
+                        {
+                            ErrorKind::NotFound
+                        } else if error_summary.contains("path/conflict") {
+                            ErrorKind::AlreadyExists
+                        } else {
+                            ErrorKind::Unexpected
+                        }
+                    }
+                    // Otherwise, we can get the error type from the response status code.
+                    _ => kind,
+                },
+                error_summary,
+            )
+            .with_context("response", format!("{parts:?}"));
 
             if retryable {
                 err = err.set_temporary();