[PR] feat(services/aliyundrive): support AliyunDrive [opendal]

["yuchanns (via GitHub)" <gi...@apache.org>]

yuchanns opened a new pull request, #4585:
URL: https://github.com/apache/opendal/pull/4585

   Will close #3415 
   
   This is just a preliminary draft to inform you that it's still under development, as it has been a while since the issue arose.
   
   I prefer to finish all of them in one pull request, so it might take some time. Thank you for your patience.


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@opendal.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] feat(services/aliyundrive): support AliyunDrive [opendal]

["yuchanns (via GitHub)" <gi...@apache.org>]

yuchanns commented on code in PR #4585:
URL: https://github.com/apache/opendal/pull/4585#discussion_r1604549227


##########
core/src/services/aliyun_drive/core.rs:
##########
@@ -0,0 +1,506 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use std::fmt::Debug;
+use std::sync::Arc;
+
+use bytes::{Buf, Bytes};
+use chrono::Utc;
+use http::header::HeaderValue;
+use http::{Method, Request};
+use serde::{Deserialize, Serialize};
+use serde_json::json;
+use tokio::sync::Mutex;
+
+use crate::raw::*;
+use crate::*;
+
+use super::error::parse_error;
+
+/// Available Aliyun Drive Type.
+#[derive(Debug, Deserialize, Default, Clone)]
+pub enum DriveType {
+    /// Use the default type of Aliyun Drive.
+    #[default]
+    Default,
+    /// Use the backup type of Aliyun Drive.
+    ///
+    /// fallback to the default type if no backup drive found.
+    Backup,
+    /// Use the resource type of Aliyun Drive.
+    ///
+    /// fallback to the default type if no resource drive found.
+    Resource,
+}
+
+pub struct AliyunDriveTokener {
+    pub drive_id: Option<String>,
+    pub access_token: Option<String>,
+    pub refresh_token: String,
+    pub expire_at: i64,
+}
+
+pub struct AliyunDriveCore {
+    pub endpoint: String,
+    pub root: String,
+    pub client_id: String,
+    pub client_secret: String,
+    pub drive_type: DriveType,
+    pub rapid_upload: bool,
+
+    pub tokener: Arc<Mutex<AliyunDriveTokener>>,
+    pub client: HttpClient,
+}
+
+impl Debug for AliyunDriveCore {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.debug_struct("AliyunDriveCore")
+            .field("root", &self.root)
+            .field("drive_type", &self.drive_type)
+            .finish_non_exhaustive()
+    }
+}
+
+impl AliyunDriveCore {
+    async fn send(&self, mut req: Request<Buffer>, token: Option<&str>) -> Result<Buffer> {
+        if req.method() == Method::POST {
+            req.headers_mut().insert(
+                "Content-Type",
+                HeaderValue::from_static("application/json;charset=UTF-8"),
+            );
+        }
+        if let Some(token) = token {
+            req.headers_mut().insert(
+                "Authorization",
+                HeaderValue::from_str(&format!("Bearer {}", token))
+                    .expect("access token must be valid header value"),
+            );
+        }
+        let res = self.client.send(req).await?;
+        if !res.status().is_success() {
+            return Err(parse_error(res).await?);
+        }
+        Ok(res.into_body())
+    }
+
+    async fn get_access_token(&self, refresh_token: &str) -> Result<Buffer> {
+        let body = serde_json::to_vec(&json!({
+            "refresh_token": refresh_token,
+            "grant_type": "refresh_token",
+            "client_id": &self.client_id,
+            "client_secret": &self.client_secret,
+        }))
+        .map_err(new_json_serialize_error)?;
+        let req = Request::post(format!("{}/oauth/access_token", self.endpoint))
+            .body(Buffer::from(bytes::Bytes::from(body)))
+            .map_err(new_request_build_error)?;
+        self.send(req, None).await
+    }
+
+    async fn get_drive_id(&self, token: Option<&str>) -> Result<Buffer> {
+        let req = Request::post(format!("{}/adrive/v1.0/user/getDriveInfo", self.endpoint))
+            .body(Buffer::new())
+            .map_err(new_request_build_error)?;
+        self.send(req, token).await
+    }
+
+    pub async fn get_token_and_drive(&self) -> Result<(Option<String>, String)> {
+        let mut tokener = self.tokener.lock().await;
+        if tokener.expire_at < Utc::now().timestamp() || tokener.access_token.is_none() {
+            let res = self.get_access_token(&tokener.refresh_token).await?;
+            let output: RefresTokenResponse =
+                serde_json::from_reader(res.reader()).map_err(new_json_deserialize_error)?;
+            tokener.access_token = Some(output.access_token);
+            tokener.expire_at = output.expires_in + Utc::now().timestamp();
+            tokener.refresh_token = output.refresh_token;
+        }
+        let Some(drive_id) = &tokener.drive_id else {
+            let res = self.get_drive_id(tokener.access_token.as_deref()).await?;
+            let output: DriveInfoResponse =
+                serde_json::from_reader(res.reader()).map_err(new_json_deserialize_error)?;
+            let drive_id = match self.drive_type {
+                DriveType::Default => output.default_drive_id,
+                DriveType::Backup => output.backup_drive_id.unwrap_or(output.default_drive_id),
+                DriveType::Resource => output.resource_drive_id.unwrap_or(output.default_drive_id),
+            };
+            tokener.drive_id = Some(drive_id.clone());
+            return Ok((tokener.access_token.clone(), drive_id));
+        };
+        Ok((tokener.access_token.clone(), drive_id.clone()))
+    }
+
+    pub async fn get_by_path(&self, path: &str) -> Result<Buffer> {
+        let file_path = build_rooted_abs_path(&self.root, path);
+        let file_path = file_path.strip_suffix('/').unwrap_or(&file_path);
+        let req = Request::post(format!(
+            "{}/adrive/v1.0/openFile/get_by_path",
+            self.endpoint
+        ));
+        let (token, drive_id) = self.get_token_and_drive().await?;
+        let body = serde_json::to_vec(&json!({
+            "drive_id": &drive_id,
+            "file_path": &file_path,
+        }))
+        .map_err(new_json_serialize_error)?;
+        let req = req
+            .body(Buffer::from(Bytes::from(body)))
+            .map_err(new_request_build_error)?;
+        self.send(req, token.as_deref()).await
+    }
+
+    pub async fn ensure_dir_exists(&self, path: &str) -> Result<String> {
+        let file_path = build_abs_path(&self.root, path.strip_suffix('/').unwrap_or(path));
+        if file_path == "/" {
+            return Ok("root".to_string());
+        }
+        let file_path = file_path.strip_suffix('/').unwrap_or(&file_path);
+        let paths = file_path.split('/').collect::<Vec<&str>>();
+        let mut parent: Option<String> = None;
+        for path in paths {
+            let res = self
+                .create(
+                    parent.as_deref(),
+                    path,
+                    CreateType::Folder,
+                    CheckNameMode::Refuse,
+                )
+                .await?;
+            let output: CreateResponse =
+                serde_json::from_reader(res.reader()).map_err(new_json_deserialize_error)?;
+            parent = Some(output.file_id);
+        }
+        Ok(parent.expect("ensure_dir_exists must succeed"))
+    }
+
+    pub async fn create_with_rapid_upload(
+        &self,
+        parent_file_id: Option<&str>,
+        name: &str,
+        typ: CreateType,
+        check_name_mode: CheckNameMode,
+        size: Option<u64>,
+        rapid_upload: Option<RapidUpload>,
+    ) -> Result<Buffer> {
+        let mut content_hash = None;
+        let mut proof_code = None;
+        let mut pre_hash = None;
+        if let Some(rapid_upload) = rapid_upload {
+            content_hash = rapid_upload.content_hash;
+            proof_code = rapid_upload.proof_code;
+            pre_hash = rapid_upload.pre_hash;
+        }
+
+        let (token, drive_id) = self.get_token_and_drive().await?;
+        let body = serde_json::to_vec(&json!({
+            "drive_id": &drive_id,
+            "parent_file_id": parent_file_id.unwrap_or("root"),
+            "name": name,
+            "type": typ,
+            "check_name_mode": check_name_mode,
+            "size": size,
+            "pre_hash": pre_hash,
+            "content_hash": content_hash,
+            "content_hash_name": content_hash.is_some().then_some("sha1"),
+            "proof_code": proof_code,
+            "proof_version": proof_code.is_some().then_some("v1"),
+        }))
+        .map_err(new_json_serialize_error)?;
+        let req = Request::post(format!("{}/adrive/v1.0/openFile/create", self.endpoint))
+            .body(Buffer::from(Bytes::from(body)))
+            .map_err(new_request_build_error)?;
+        self.send(req, token.as_deref()).await
+    }
+
+    pub async fn create(
+        &self,
+        parent_file_id: Option<&str>,
+        name: &str,
+        typ: CreateType,
+        check_name_mode: CheckNameMode,
+    ) -> Result<Buffer> {
+        self.create_with_rapid_upload(parent_file_id, name, typ, check_name_mode, None, None)
+            .await
+    }
+
+    pub async fn get_download_url(&self, file_id: &str) -> Result<String> {
+        let (token, drive_id) = self.get_token_and_drive().await?;
+        let body = serde_json::to_vec(&json!({
+            "drive_id": &drive_id,
+            "file_id": file_id,
+        }))
+        .map_err(new_json_serialize_error)?;
+        let req = Request::post(format!(
+            "{}/adrive/v1.0/openFile/getDownloadUrl",
+            self.endpoint
+        ))
+        .body(Buffer::from(Bytes::from(body)))
+        .map_err(new_request_build_error)?;
+        let res = self.send(req, token.as_deref()).await?;
+        let output: GetDownloadUrlResponse =
+            serde_json::from_reader(res.reader()).map_err(new_json_serialize_error)?;
+        Ok(output.url)
+    }
+
+    pub async fn move_path(&self, file_id: &str, to_parent_file_id: &str) -> Result<()> {
+        let (token, drive_id) = self.get_token_and_drive().await?;
+        let body = serde_json::to_vec(&json!({
+            "drive_id": &drive_id,
+            "file_id": file_id,
+            "to_parent_file_id": to_parent_file_id,
+            "check_name_mode": CheckNameMode::AutoRename,
+        }))
+        .map_err(new_json_serialize_error)?;
+        let req = Request::post(format!("{}/adrive/v1.0/openFile/move", self.endpoint))
+            .body(Buffer::from(Bytes::from(body)))
+            .map_err(new_request_build_error)?;
+        self.send(req, token.as_deref()).await?;
+        Ok(())
+    }
+
+    pub async fn update_path(&self, file_id: &str, name: &str) -> Result<()> {
+        let (token, drive_id) = self.get_token_and_drive().await?;
+        let body = serde_json::to_vec(&json!({

Review Comment:
   SGTM



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@opendal.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] feat(services/aliyun-drive): support AliyunDrive [opendal]

["Xuanwo (via GitHub)" <gi...@apache.org>]

Xuanwo merged PR #4585:
URL: https://github.com/apache/opendal/pull/4585


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@opendal.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] feat(services/aliyun-drive): support AliyunDrive [opendal]

["Xuanwo (via GitHub)" <gi...@apache.org>]

Xuanwo commented on code in PR #4585:
URL: https://github.com/apache/opendal/pull/4585#discussion_r1605032702


##########
.github/services/aliyun_drive/aliyun_drive/action.yml:
##########


Review Comment:
   Please rename this file as `disabled_action.yml`. We need some time to setup the account.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@opendal.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] feat(services/aliyun-drive): support AliyunDrive [opendal]

["imWildCat (via GitHub)" <gi...@apache.org>]

imWildCat commented on PR #4585:
URL: https://github.com/apache/opendal/pull/4585#issuecomment-2166698100

   Trying to use this in production but this implementation is too different from that of OneDrive and Google Drive.
   
   The major question is that: Why this OpenDAL implement requires refresh_token and client_id?
   
   The AliyunDrive API supports OAuth. IMHO, we can just send an access_token to the operator and let the host app handle the auth part.


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@opendal.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] feat(services/aliyun-drive): support AliyunDrive [opendal]

["imWildCat (via GitHub)" <gi...@apache.org>]

imWildCat commented on PR #4585:
URL: https://github.com/apache/opendal/pull/4585#issuecomment-2118842154

   thank you!


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@opendal.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] feat(services/aliyundrive): support AliyunDrive [opendal]

["Xuanwo (via GitHub)" <gi...@apache.org>]

Xuanwo commented on code in PR #4585:
URL: https://github.com/apache/opendal/pull/4585#discussion_r1604425874


##########
core/src/services/aliyun_drive/core.rs:
##########
@@ -0,0 +1,506 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use std::fmt::Debug;
+use std::sync::Arc;
+
+use bytes::{Buf, Bytes};
+use chrono::Utc;
+use http::header::HeaderValue;
+use http::{Method, Request};
+use serde::{Deserialize, Serialize};
+use serde_json::json;
+use tokio::sync::Mutex;
+
+use crate::raw::*;
+use crate::*;
+
+use super::error::parse_error;
+
+/// Available Aliyun Drive Type.
+#[derive(Debug, Deserialize, Default, Clone)]
+pub enum DriveType {
+    /// Use the default type of Aliyun Drive.
+    #[default]
+    Default,
+    /// Use the backup type of Aliyun Drive.
+    ///
+    /// fallback to the default type if no backup drive found.
+    Backup,
+    /// Use the resource type of Aliyun Drive.
+    ///
+    /// fallback to the default type if no resource drive found.
+    Resource,
+}
+
+pub struct AliyunDriveTokener {
+    pub drive_id: Option<String>,
+    pub access_token: Option<String>,
+    pub refresh_token: String,
+    pub expire_at: i64,
+}
+
+pub struct AliyunDriveCore {
+    pub endpoint: String,
+    pub root: String,
+    pub client_id: String,
+    pub client_secret: String,
+    pub drive_type: DriveType,
+    pub rapid_upload: bool,
+
+    pub tokener: Arc<Mutex<AliyunDriveTokener>>,
+    pub client: HttpClient,
+}
+
+impl Debug for AliyunDriveCore {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.debug_struct("AliyunDriveCore")
+            .field("root", &self.root)
+            .field("drive_type", &self.drive_type)
+            .finish_non_exhaustive()
+    }
+}
+
+impl AliyunDriveCore {
+    async fn send(&self, mut req: Request<Buffer>, token: Option<&str>) -> Result<Buffer> {
+        if req.method() == Method::POST {
+            req.headers_mut().insert(
+                "Content-Type",
+                HeaderValue::from_static("application/json;charset=UTF-8"),
+            );
+        }
+        if let Some(token) = token {
+            req.headers_mut().insert(
+                "Authorization",
+                HeaderValue::from_str(&format!("Bearer {}", token))
+                    .expect("access token must be valid header value"),
+            );
+        }
+        let res = self.client.send(req).await?;
+        if !res.status().is_success() {
+            return Err(parse_error(res).await?);
+        }
+        Ok(res.into_body())
+    }
+
+    async fn get_access_token(&self, refresh_token: &str) -> Result<Buffer> {
+        let body = serde_json::to_vec(&json!({
+            "refresh_token": refresh_token,
+            "grant_type": "refresh_token",
+            "client_id": &self.client_id,
+            "client_secret": &self.client_secret,
+        }))
+        .map_err(new_json_serialize_error)?;
+        let req = Request::post(format!("{}/oauth/access_token", self.endpoint))
+            .body(Buffer::from(bytes::Bytes::from(body)))
+            .map_err(new_request_build_error)?;
+        self.send(req, None).await
+    }
+
+    async fn get_drive_id(&self, token: Option<&str>) -> Result<Buffer> {
+        let req = Request::post(format!("{}/adrive/v1.0/user/getDriveInfo", self.endpoint))
+            .body(Buffer::new())
+            .map_err(new_request_build_error)?;
+        self.send(req, token).await
+    }
+
+    pub async fn get_token_and_drive(&self) -> Result<(Option<String>, String)> {
+        let mut tokener = self.tokener.lock().await;
+        if tokener.expire_at < Utc::now().timestamp() || tokener.access_token.is_none() {
+            let res = self.get_access_token(&tokener.refresh_token).await?;
+            let output: RefresTokenResponse =
+                serde_json::from_reader(res.reader()).map_err(new_json_deserialize_error)?;
+            tokener.access_token = Some(output.access_token);
+            tokener.expire_at = output.expires_in + Utc::now().timestamp();
+            tokener.refresh_token = output.refresh_token;
+        }
+        let Some(drive_id) = &tokener.drive_id else {
+            let res = self.get_drive_id(tokener.access_token.as_deref()).await?;
+            let output: DriveInfoResponse =
+                serde_json::from_reader(res.reader()).map_err(new_json_deserialize_error)?;
+            let drive_id = match self.drive_type {
+                DriveType::Default => output.default_drive_id,
+                DriveType::Backup => output.backup_drive_id.unwrap_or(output.default_drive_id),
+                DriveType::Resource => output.resource_drive_id.unwrap_or(output.default_drive_id),
+            };
+            tokener.drive_id = Some(drive_id.clone());
+            return Ok((tokener.access_token.clone(), drive_id));
+        };
+        Ok((tokener.access_token.clone(), drive_id.clone()))
+    }
+
+    pub async fn get_by_path(&self, path: &str) -> Result<Buffer> {
+        let file_path = build_rooted_abs_path(&self.root, path);
+        let file_path = file_path.strip_suffix('/').unwrap_or(&file_path);
+        let req = Request::post(format!(
+            "{}/adrive/v1.0/openFile/get_by_path",
+            self.endpoint
+        ));
+        let (token, drive_id) = self.get_token_and_drive().await?;
+        let body = serde_json::to_vec(&json!({
+            "drive_id": &drive_id,
+            "file_path": &file_path,
+        }))
+        .map_err(new_json_serialize_error)?;
+        let req = req
+            .body(Buffer::from(Bytes::from(body)))
+            .map_err(new_request_build_error)?;
+        self.send(req, token.as_deref()).await
+    }
+
+    pub async fn ensure_dir_exists(&self, path: &str) -> Result<String> {
+        let file_path = build_abs_path(&self.root, path.strip_suffix('/').unwrap_or(path));
+        if file_path == "/" {
+            return Ok("root".to_string());
+        }
+        let file_path = file_path.strip_suffix('/').unwrap_or(&file_path);
+        let paths = file_path.split('/').collect::<Vec<&str>>();
+        let mut parent: Option<String> = None;
+        for path in paths {
+            let res = self
+                .create(
+                    parent.as_deref(),
+                    path,
+                    CreateType::Folder,
+                    CheckNameMode::Refuse,
+                )
+                .await?;
+            let output: CreateResponse =
+                serde_json::from_reader(res.reader()).map_err(new_json_deserialize_error)?;
+            parent = Some(output.file_id);
+        }
+        Ok(parent.expect("ensure_dir_exists must succeed"))
+    }
+
+    pub async fn create_with_rapid_upload(
+        &self,
+        parent_file_id: Option<&str>,
+        name: &str,
+        typ: CreateType,
+        check_name_mode: CheckNameMode,
+        size: Option<u64>,
+        rapid_upload: Option<RapidUpload>,
+    ) -> Result<Buffer> {
+        let mut content_hash = None;
+        let mut proof_code = None;
+        let mut pre_hash = None;
+        if let Some(rapid_upload) = rapid_upload {
+            content_hash = rapid_upload.content_hash;
+            proof_code = rapid_upload.proof_code;
+            pre_hash = rapid_upload.pre_hash;
+        }
+
+        let (token, drive_id) = self.get_token_and_drive().await?;
+        let body = serde_json::to_vec(&json!({
+            "drive_id": &drive_id,
+            "parent_file_id": parent_file_id.unwrap_or("root"),
+            "name": name,
+            "type": typ,
+            "check_name_mode": check_name_mode,
+            "size": size,
+            "pre_hash": pre_hash,
+            "content_hash": content_hash,
+            "content_hash_name": content_hash.is_some().then_some("sha1"),
+            "proof_code": proof_code,
+            "proof_version": proof_code.is_some().then_some("v1"),
+        }))
+        .map_err(new_json_serialize_error)?;
+        let req = Request::post(format!("{}/adrive/v1.0/openFile/create", self.endpoint))
+            .body(Buffer::from(Bytes::from(body)))
+            .map_err(new_request_build_error)?;
+        self.send(req, token.as_deref()).await
+    }
+
+    pub async fn create(
+        &self,
+        parent_file_id: Option<&str>,
+        name: &str,
+        typ: CreateType,
+        check_name_mode: CheckNameMode,
+    ) -> Result<Buffer> {
+        self.create_with_rapid_upload(parent_file_id, name, typ, check_name_mode, None, None)
+            .await
+    }
+
+    pub async fn get_download_url(&self, file_id: &str) -> Result<String> {
+        let (token, drive_id) = self.get_token_and_drive().await?;
+        let body = serde_json::to_vec(&json!({
+            "drive_id": &drive_id,
+            "file_id": file_id,
+        }))
+        .map_err(new_json_serialize_error)?;
+        let req = Request::post(format!(
+            "{}/adrive/v1.0/openFile/getDownloadUrl",
+            self.endpoint
+        ))
+        .body(Buffer::from(Bytes::from(body)))
+        .map_err(new_request_build_error)?;
+        let res = self.send(req, token.as_deref()).await?;
+        let output: GetDownloadUrlResponse =
+            serde_json::from_reader(res.reader()).map_err(new_json_serialize_error)?;
+        Ok(output.url)
+    }
+
+    pub async fn move_path(&self, file_id: &str, to_parent_file_id: &str) -> Result<()> {
+        let (token, drive_id) = self.get_token_and_drive().await?;
+        let body = serde_json::to_vec(&json!({
+            "drive_id": &drive_id,
+            "file_id": file_id,
+            "to_parent_file_id": to_parent_file_id,
+            "check_name_mode": CheckNameMode::AutoRename,
+        }))
+        .map_err(new_json_serialize_error)?;
+        let req = Request::post(format!("{}/adrive/v1.0/openFile/move", self.endpoint))
+            .body(Buffer::from(Bytes::from(body)))
+            .map_err(new_request_build_error)?;
+        self.send(req, token.as_deref()).await?;
+        Ok(())
+    }
+
+    pub async fn update_path(&self, file_id: &str, name: &str) -> Result<()> {
+        let (token, drive_id) = self.get_token_and_drive().await?;
+        let body = serde_json::to_vec(&json!({

Review Comment:
   > It looks like there's a lot of `json!()` being used in the repo. Should I keep the consistency?
   
   Yep, I want to remove them all. We can track this task in seperate issues.
   
   > Also, are you suggesting that all instances of `json!()` should be removed from this file?
   
   Yes, we can at least avoid adding new `json!()`.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@opendal.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] feat(services/aliyundrive): support AliyunDrive [opendal]

["yuchanns (via GitHub)" <gi...@apache.org>]

yuchanns commented on code in PR #4585:
URL: https://github.com/apache/opendal/pull/4585#discussion_r1604418561


##########
core/src/services/aliyun_drive/core.rs:
##########
@@ -0,0 +1,506 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use std::fmt::Debug;
+use std::sync::Arc;
+
+use bytes::{Buf, Bytes};
+use chrono::Utc;
+use http::header::HeaderValue;
+use http::{Method, Request};
+use serde::{Deserialize, Serialize};
+use serde_json::json;
+use tokio::sync::Mutex;
+
+use crate::raw::*;
+use crate::*;
+
+use super::error::parse_error;
+
+/// Available Aliyun Drive Type.
+#[derive(Debug, Deserialize, Default, Clone)]
+pub enum DriveType {
+    /// Use the default type of Aliyun Drive.
+    #[default]
+    Default,
+    /// Use the backup type of Aliyun Drive.
+    ///
+    /// fallback to the default type if no backup drive found.
+    Backup,
+    /// Use the resource type of Aliyun Drive.
+    ///
+    /// fallback to the default type if no resource drive found.
+    Resource,
+}
+
+pub struct AliyunDriveTokener {
+    pub drive_id: Option<String>,
+    pub access_token: Option<String>,
+    pub refresh_token: String,
+    pub expire_at: i64,
+}
+
+pub struct AliyunDriveCore {
+    pub endpoint: String,
+    pub root: String,
+    pub client_id: String,
+    pub client_secret: String,
+    pub drive_type: DriveType,
+    pub rapid_upload: bool,
+
+    pub tokener: Arc<Mutex<AliyunDriveTokener>>,
+    pub client: HttpClient,
+}
+
+impl Debug for AliyunDriveCore {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.debug_struct("AliyunDriveCore")
+            .field("root", &self.root)
+            .field("drive_type", &self.drive_type)
+            .finish_non_exhaustive()
+    }
+}
+
+impl AliyunDriveCore {
+    async fn send(&self, mut req: Request<Buffer>, token: Option<&str>) -> Result<Buffer> {
+        if req.method() == Method::POST {
+            req.headers_mut().insert(
+                "Content-Type",
+                HeaderValue::from_static("application/json;charset=UTF-8"),
+            );
+        }
+        if let Some(token) = token {
+            req.headers_mut().insert(
+                "Authorization",
+                HeaderValue::from_str(&format!("Bearer {}", token))
+                    .expect("access token must be valid header value"),
+            );
+        }
+        let res = self.client.send(req).await?;
+        if !res.status().is_success() {
+            return Err(parse_error(res).await?);
+        }
+        Ok(res.into_body())
+    }
+
+    async fn get_access_token(&self, refresh_token: &str) -> Result<Buffer> {
+        let body = serde_json::to_vec(&json!({
+            "refresh_token": refresh_token,
+            "grant_type": "refresh_token",
+            "client_id": &self.client_id,
+            "client_secret": &self.client_secret,
+        }))
+        .map_err(new_json_serialize_error)?;
+        let req = Request::post(format!("{}/oauth/access_token", self.endpoint))
+            .body(Buffer::from(bytes::Bytes::from(body)))
+            .map_err(new_request_build_error)?;
+        self.send(req, None).await
+    }
+
+    async fn get_drive_id(&self, token: Option<&str>) -> Result<Buffer> {
+        let req = Request::post(format!("{}/adrive/v1.0/user/getDriveInfo", self.endpoint))
+            .body(Buffer::new())
+            .map_err(new_request_build_error)?;
+        self.send(req, token).await
+    }
+
+    pub async fn get_token_and_drive(&self) -> Result<(Option<String>, String)> {
+        let mut tokener = self.tokener.lock().await;
+        if tokener.expire_at < Utc::now().timestamp() || tokener.access_token.is_none() {
+            let res = self.get_access_token(&tokener.refresh_token).await?;
+            let output: RefresTokenResponse =
+                serde_json::from_reader(res.reader()).map_err(new_json_deserialize_error)?;
+            tokener.access_token = Some(output.access_token);
+            tokener.expire_at = output.expires_in + Utc::now().timestamp();
+            tokener.refresh_token = output.refresh_token;
+        }
+        let Some(drive_id) = &tokener.drive_id else {
+            let res = self.get_drive_id(tokener.access_token.as_deref()).await?;
+            let output: DriveInfoResponse =
+                serde_json::from_reader(res.reader()).map_err(new_json_deserialize_error)?;
+            let drive_id = match self.drive_type {
+                DriveType::Default => output.default_drive_id,
+                DriveType::Backup => output.backup_drive_id.unwrap_or(output.default_drive_id),
+                DriveType::Resource => output.resource_drive_id.unwrap_or(output.default_drive_id),
+            };
+            tokener.drive_id = Some(drive_id.clone());
+            return Ok((tokener.access_token.clone(), drive_id));
+        };
+        Ok((tokener.access_token.clone(), drive_id.clone()))
+    }
+
+    pub async fn get_by_path(&self, path: &str) -> Result<Buffer> {
+        let file_path = build_rooted_abs_path(&self.root, path);
+        let file_path = file_path.strip_suffix('/').unwrap_or(&file_path);
+        let req = Request::post(format!(
+            "{}/adrive/v1.0/openFile/get_by_path",
+            self.endpoint
+        ));
+        let (token, drive_id) = self.get_token_and_drive().await?;
+        let body = serde_json::to_vec(&json!({
+            "drive_id": &drive_id,
+            "file_path": &file_path,
+        }))
+        .map_err(new_json_serialize_error)?;
+        let req = req
+            .body(Buffer::from(Bytes::from(body)))
+            .map_err(new_request_build_error)?;
+        self.send(req, token.as_deref()).await
+    }
+
+    pub async fn ensure_dir_exists(&self, path: &str) -> Result<String> {
+        let file_path = build_abs_path(&self.root, path.strip_suffix('/').unwrap_or(path));
+        if file_path == "/" {
+            return Ok("root".to_string());
+        }
+        let file_path = file_path.strip_suffix('/').unwrap_or(&file_path);
+        let paths = file_path.split('/').collect::<Vec<&str>>();
+        let mut parent: Option<String> = None;
+        for path in paths {
+            let res = self
+                .create(
+                    parent.as_deref(),
+                    path,
+                    CreateType::Folder,
+                    CheckNameMode::Refuse,
+                )
+                .await?;
+            let output: CreateResponse =
+                serde_json::from_reader(res.reader()).map_err(new_json_deserialize_error)?;
+            parent = Some(output.file_id);
+        }
+        Ok(parent.expect("ensure_dir_exists must succeed"))
+    }
+
+    pub async fn create_with_rapid_upload(
+        &self,
+        parent_file_id: Option<&str>,
+        name: &str,
+        typ: CreateType,
+        check_name_mode: CheckNameMode,
+        size: Option<u64>,
+        rapid_upload: Option<RapidUpload>,
+    ) -> Result<Buffer> {
+        let mut content_hash = None;
+        let mut proof_code = None;
+        let mut pre_hash = None;
+        if let Some(rapid_upload) = rapid_upload {
+            content_hash = rapid_upload.content_hash;
+            proof_code = rapid_upload.proof_code;
+            pre_hash = rapid_upload.pre_hash;
+        }
+
+        let (token, drive_id) = self.get_token_and_drive().await?;
+        let body = serde_json::to_vec(&json!({
+            "drive_id": &drive_id,
+            "parent_file_id": parent_file_id.unwrap_or("root"),
+            "name": name,
+            "type": typ,
+            "check_name_mode": check_name_mode,
+            "size": size,
+            "pre_hash": pre_hash,
+            "content_hash": content_hash,
+            "content_hash_name": content_hash.is_some().then_some("sha1"),
+            "proof_code": proof_code,
+            "proof_version": proof_code.is_some().then_some("v1"),
+        }))
+        .map_err(new_json_serialize_error)?;
+        let req = Request::post(format!("{}/adrive/v1.0/openFile/create", self.endpoint))
+            .body(Buffer::from(Bytes::from(body)))
+            .map_err(new_request_build_error)?;
+        self.send(req, token.as_deref()).await
+    }
+
+    pub async fn create(
+        &self,
+        parent_file_id: Option<&str>,
+        name: &str,
+        typ: CreateType,
+        check_name_mode: CheckNameMode,
+    ) -> Result<Buffer> {
+        self.create_with_rapid_upload(parent_file_id, name, typ, check_name_mode, None, None)
+            .await
+    }
+
+    pub async fn get_download_url(&self, file_id: &str) -> Result<String> {
+        let (token, drive_id) = self.get_token_and_drive().await?;
+        let body = serde_json::to_vec(&json!({
+            "drive_id": &drive_id,
+            "file_id": file_id,
+        }))
+        .map_err(new_json_serialize_error)?;
+        let req = Request::post(format!(
+            "{}/adrive/v1.0/openFile/getDownloadUrl",
+            self.endpoint
+        ))
+        .body(Buffer::from(Bytes::from(body)))
+        .map_err(new_request_build_error)?;
+        let res = self.send(req, token.as_deref()).await?;
+        let output: GetDownloadUrlResponse =
+            serde_json::from_reader(res.reader()).map_err(new_json_serialize_error)?;
+        Ok(output.url)
+    }
+
+    pub async fn move_path(&self, file_id: &str, to_parent_file_id: &str) -> Result<()> {
+        let (token, drive_id) = self.get_token_and_drive().await?;
+        let body = serde_json::to_vec(&json!({
+            "drive_id": &drive_id,
+            "file_id": file_id,
+            "to_parent_file_id": to_parent_file_id,
+            "check_name_mode": CheckNameMode::AutoRename,
+        }))
+        .map_err(new_json_serialize_error)?;
+        let req = Request::post(format!("{}/adrive/v1.0/openFile/move", self.endpoint))
+            .body(Buffer::from(Bytes::from(body)))
+            .map_err(new_request_build_error)?;
+        self.send(req, token.as_deref()).await?;
+        Ok(())
+    }
+
+    pub async fn update_path(&self, file_id: &str, name: &str) -> Result<()> {
+        let (token, drive_id) = self.get_token_and_drive().await?;
+        let body = serde_json::to_vec(&json!({

Review Comment:
   It looks like there's a lot of `json!()` being used in the repo. Should I keep the consistency?
   Also, are you suggesting that all instances of `json!()` should be removed from this file?



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@opendal.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] feat(services/aliyundrive): support AliyunDrive [opendal]

["yuchanns (via GitHub)" <gi...@apache.org>]

yuchanns commented on PR #4585:
URL: https://github.com/apache/opendal/pull/4585#issuecomment-2115577199

   Sure, I'll fix them later today.


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@opendal.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] feat(services/aliyundrive): support AliyunDrive [opendal]

["suyanhanx (via GitHub)" <gi...@apache.org>]

suyanhanx commented on code in PR #4585:
URL: https://github.com/apache/opendal/pull/4585#discussion_r1603480317


##########
core/src/services/aliyun_drive/backend.rs:
##########
@@ -0,0 +1,457 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use std::fmt::Debug;
+use std::fmt::Formatter;
+use std::sync::Arc;
+
+use bytes::Buf;
+use chrono::Utc;
+use log::debug;
+use serde::Deserialize;
+use tokio::sync::Mutex;
+
+use super::core::*;
+use super::lister::AliyunDriveLister;
+use super::lister::AliyunDriveParent;
+use super::reader::AliyunDriveReader;
+use super::writer::{AliyunDriveWriter, AliyunDriveWriters};
+use crate::raw::*;
+use crate::*;
+
+/// Aliyun Drive services support.
+#[derive(Default, Deserialize)]
+#[serde(default)]
+#[non_exhaustive]
+pub struct AliyunDriveConfig {
+    /// root of this backend.
+    ///
+    /// All operations will happen under this root.
+    ///
+    /// default to `/` if not set.
+    pub root: Option<String>,
+    /// client_id of this backend.
+    ///
+    /// required.
+    pub client_id: String,
+    /// client_secret of this backend.
+    ///
+    /// required.
+    pub client_secret: String,
+    /// refresh_token of this backend.
+    ///
+    /// required.
+    pub refresh_token: String,
+    /// drive_type of this backend.
+    ///
+    /// All operations will happen under this type of drive.
+    ///
+    /// default to DriveType::Default if not set.
+    pub drive_type: DriveType,
+    /// rapid_upload of this backend.

Review Comment:
   The so-called "rapid upload" or "秒传" skips the process of uploading files already in the cloud by hashing the content.
   



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@opendal.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] feat(services/aliyun-drive): support AliyunDrive [opendal]

["Xuanwo (via GitHub)" <gi...@apache.org>]

Xuanwo commented on PR #4585:
URL: https://github.com/apache/opendal/pull/4585#issuecomment-2117553033

   > AliyunDrive has a feature called Rapid Upload, but it is difficult to implement completely because the `MultipartWriter` trait design isn't well compatible with this feature at present. Therefore, I have only implemented it on the `write_once` method.
   
   Hi, it's not a requirement to implement `MultipartWriter`. Feel free to implement `oio::Write` on you own.


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@opendal.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] feat(services/aliyun-drive): support AliyunDrive [opendal]

["suyanhanx (via GitHub)" <gi...@apache.org>]

suyanhanx commented on PR #4585:
URL: https://github.com/apache/opendal/pull/4585#issuecomment-2117865073

   I'll set up the test credentials for this service.


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@opendal.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] feat(services/aliyun-drive): support AliyunDrive [opendal]

["suyanhanx (via GitHub)" <gi...@apache.org>]

suyanhanx commented on PR #4585:
URL: https://github.com/apache/opendal/pull/4585#issuecomment-2117858358

   Great work! Thank you!


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@opendal.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] feat(services/aliyundrive): support AliyunDrive [opendal]

["Xuanwo (via GitHub)" <gi...@apache.org>]

Xuanwo commented on code in PR #4585:
URL: https://github.com/apache/opendal/pull/4585#discussion_r1603141538


##########
core/src/services/aliyun_drive/backend.rs:
##########
@@ -0,0 +1,457 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use std::fmt::Debug;
+use std::fmt::Formatter;
+use std::sync::Arc;
+
+use bytes::Buf;
+use chrono::Utc;
+use log::debug;
+use serde::Deserialize;
+use tokio::sync::Mutex;
+
+use super::core::*;
+use super::lister::AliyunDriveLister;
+use super::lister::AliyunDriveParent;
+use super::reader::AliyunDriveReader;
+use super::writer::{AliyunDriveWriter, AliyunDriveWriters};
+use crate::raw::*;
+use crate::*;
+
+/// Aliyun Drive services support.
+#[derive(Default, Deserialize)]
+#[serde(default)]
+#[non_exhaustive]
+pub struct AliyunDriveConfig {
+    /// root of this backend.
+    ///
+    /// All operations will happen under this root.
+    ///
+    /// default to `/` if not set.
+    pub root: Option<String>,
+    /// client_id of this backend.
+    ///
+    /// required.
+    pub client_id: String,
+    /// client_secret of this backend.
+    ///
+    /// required.
+    pub client_secret: String,
+    /// refresh_token of this backend.
+    ///
+    /// required.
+    pub refresh_token: String,
+    /// drive_type of this backend.
+    ///
+    /// All operations will happen under this type of drive.
+    ///
+    /// default to DriveType::Default if not set.
+    pub drive_type: DriveType,

Review Comment:
   It's better to use `String` in config instead of an new type to make our config parser much easier to implement.



##########
core/src/services/aliyun_drive/backend.rs:
##########
@@ -0,0 +1,457 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use std::fmt::Debug;
+use std::fmt::Formatter;
+use std::sync::Arc;
+
+use bytes::Buf;
+use chrono::Utc;
+use log::debug;
+use serde::Deserialize;
+use tokio::sync::Mutex;
+
+use super::core::*;
+use super::lister::AliyunDriveLister;
+use super::lister::AliyunDriveParent;
+use super::reader::AliyunDriveReader;
+use super::writer::{AliyunDriveWriter, AliyunDriveWriters};
+use crate::raw::*;
+use crate::*;
+
+/// Aliyun Drive services support.
+#[derive(Default, Deserialize)]
+#[serde(default)]
+#[non_exhaustive]
+pub struct AliyunDriveConfig {
+    /// root of this backend.
+    ///
+    /// All operations will happen under this root.
+    ///
+    /// default to `/` if not set.
+    pub root: Option<String>,
+    /// client_id of this backend.
+    ///
+    /// required.
+    pub client_id: String,
+    /// client_secret of this backend.
+    ///
+    /// required.
+    pub client_secret: String,
+    /// refresh_token of this backend.
+    ///
+    /// required.
+    pub refresh_token: String,
+    /// drive_type of this backend.
+    ///
+    /// All operations will happen under this type of drive.
+    ///
+    /// default to DriveType::Default if not set.
+    pub drive_type: DriveType,
+    /// rapid_upload of this backend.
+    ///
+    /// Only works under the write_once operation.
+    pub rapid_upload: bool,
+}
+
+impl Debug for AliyunDriveConfig {
+    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+        let mut d = f.debug_struct("AliyunDriveConfig");
+
+        d.field("root", &self.root)
+            .field("drive_type", &self.drive_type);
+
+        d.finish_non_exhaustive()
+    }
+}
+
+#[doc = include_str!("docs.md")]
+#[derive(Default)]
+pub struct AliyunDriveBuilder {
+    config: AliyunDriveConfig,
+
+    http_client: Option<HttpClient>,
+}
+
+impl Debug for AliyunDriveBuilder {
+    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+        let mut d = f.debug_struct("AliyunDriveBuilder");
+
+        d.field("config", &self.config);
+        d.finish_non_exhaustive()
+    }
+}
+
+impl AliyunDriveBuilder {
+    /// Set root of this backend.
+    ///
+    /// All operations will happen under this root.
+    pub fn root(&mut self, root: &str) -> &mut Self {
+        self.config.root = if root.is_empty() {
+            None
+        } else {
+            Some(root.to_string())
+        };
+
+        self
+    }
+
+    /// Set client_id of this backend.
+    pub fn client_id(&mut self, client_id: &str) -> &mut Self {
+        self.config.client_id = client_id.to_string();
+
+        self
+    }
+
+    /// Set client_secret of this backend.
+    pub fn client_secret(&mut self, client_secret: &str) -> &mut Self {
+        self.config.client_secret = client_secret.to_string();
+
+        self
+    }
+
+    /// Set refresh_token of this backend.
+    pub fn refresh_token(&mut self, refresh_token: &str) -> &mut Self {
+        self.config.refresh_token = refresh_token.to_string();
+
+        self
+    }
+
+    /// Set drive_type of this backend.
+    pub fn drive_type(&mut self, drive_type: DriveType) -> &mut Self {
+        self.config.drive_type = drive_type;
+
+        self
+    }
+
+    /// Set rapid_upload of this backend.
+    pub fn rapid_upload(&mut self, rapid_upload: bool) -> &mut Self {
+        self.config.rapid_upload = rapid_upload;
+
+        self
+    }
+
+    /// Specify the http client that used by this service.
+    ///
+    /// # Notes
+    ///
+    /// This API is part of OpenDAL's Raw API. `HttpClient` could be changed
+    /// during minor updates.
+    pub fn http_client(&mut self, client: HttpClient) -> &mut Self {
+        self.http_client = Some(client);
+        self
+    }
+}
+
+impl Builder for AliyunDriveBuilder {
+    const SCHEME: Scheme = Scheme::AliyunDrive;
+
+    type Accessor = AliyunDriveBackend;
+
+    fn from_map(map: std::collections::HashMap<String, String>) -> Self {
+        let config = AliyunDriveConfig::deserialize(ConfigDeserializer::new(map))
+            .expect("config deserialize must succeed");
+        AliyunDriveBuilder {
+            config,
+
+            http_client: None,
+        }
+    }
+
+    fn build(&mut self) -> Result<Self::Accessor> {
+        debug!("backend build started: {:?}", &self);
+
+        let root = normalize_root(&self.config.root.clone().unwrap_or_default());
+        debug!("backend use root {}", &root);
+
+        let client = if let Some(client) = self.http_client.take() {
+            client
+        } else {
+            HttpClient::new().map_err(|err| {
+                err.with_operation("Builder::build")
+                    .with_context("service", Scheme::AliyunDrive)
+            })?
+        };
+
+        let client_id = self.config.client_id.clone();
+        if client_id.is_empty() {
+            return Err(
+                Error::new(ErrorKind::ConfigInvalid, "client_id is missing.")
+                    .with_operation("Builder::build")
+                    .with_context("service", Scheme::AliyunDrive),
+            );
+        }
+
+        let client_secret = self.config.client_secret.clone();
+        if client_secret.is_empty() {
+            return Err(
+                Error::new(ErrorKind::ConfigInvalid, "client_secret is missing.")
+                    .with_operation("Builder::build")
+                    .with_context("service", Scheme::AliyunDrive),
+            );
+        }
+
+        let refresh_token = self.config.refresh_token.clone();
+        if refresh_token.is_empty() {
+            return Err(
+                Error::new(ErrorKind::ConfigInvalid, "refresh_token is missing.")
+                    .with_operation("Builder::build")
+                    .with_context("service", Scheme::AliyunDrive),
+            );
+        }
+
+        let drive_type = self.config.drive_type.clone();
+        println!("backend use drive_type {:?}", drive_type);
+
+        let rapid_upload = self.config.rapid_upload;
+        debug!("backend use rapid_upload {}", rapid_upload);
+
+        Ok(AliyunDriveBackend {
+            core: Arc::new(AliyunDriveCore {
+                endpoint: "https://openapi.alipan.com".to_string(),
+                root,
+                client_id,
+                client_secret,
+                drive_type,
+                rapid_upload,
+                tokener: Arc::new(Mutex::new(AliyunDriveTokener {

Review Comment:
   `tokener` is unusual in opendal project, how about using `signer`?



##########
core/src/services/aliyun_drive/core.rs:
##########
@@ -0,0 +1,506 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use std::fmt::Debug;
+use std::sync::Arc;
+
+use bytes::{Buf, Bytes};
+use chrono::Utc;
+use http::header::HeaderValue;
+use http::{Method, Request};
+use serde::{Deserialize, Serialize};
+use serde_json::json;
+use tokio::sync::Mutex;
+
+use crate::raw::*;
+use crate::*;
+
+use super::error::parse_error;
+
+/// Available Aliyun Drive Type.
+#[derive(Debug, Deserialize, Default, Clone)]
+pub enum DriveType {
+    /// Use the default type of Aliyun Drive.
+    #[default]
+    Default,
+    /// Use the backup type of Aliyun Drive.
+    ///
+    /// fallback to the default type if no backup drive found.
+    Backup,
+    /// Use the resource type of Aliyun Drive.
+    ///
+    /// fallback to the default type if no resource drive found.
+    Resource,
+}
+
+pub struct AliyunDriveTokener {
+    pub drive_id: Option<String>,
+    pub access_token: Option<String>,
+    pub refresh_token: String,
+    pub expire_at: i64,
+}
+
+pub struct AliyunDriveCore {
+    pub endpoint: String,
+    pub root: String,
+    pub client_id: String,
+    pub client_secret: String,
+    pub drive_type: DriveType,
+    pub rapid_upload: bool,
+
+    pub tokener: Arc<Mutex<AliyunDriveTokener>>,
+    pub client: HttpClient,
+}
+
+impl Debug for AliyunDriveCore {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.debug_struct("AliyunDriveCore")
+            .field("root", &self.root)
+            .field("drive_type", &self.drive_type)
+            .finish_non_exhaustive()
+    }
+}
+
+impl AliyunDriveCore {
+    async fn send(&self, mut req: Request<Buffer>, token: Option<&str>) -> Result<Buffer> {
+        if req.method() == Method::POST {
+            req.headers_mut().insert(
+                "Content-Type",
+                HeaderValue::from_static("application/json;charset=UTF-8"),
+            );
+        }
+        if let Some(token) = token {
+            req.headers_mut().insert(
+                "Authorization",
+                HeaderValue::from_str(&format!("Bearer {}", token))

Review Comment:
   Please use [format_authorization_by_bearer](https://opendal.apache.org/docs/rust/opendal/raw/fn.format_authorization_by_bearer.html)



##########
core/src/services/aliyun_drive/backend.rs:
##########
@@ -0,0 +1,457 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use std::fmt::Debug;
+use std::fmt::Formatter;
+use std::sync::Arc;
+
+use bytes::Buf;
+use chrono::Utc;
+use log::debug;
+use serde::Deserialize;
+use tokio::sync::Mutex;
+
+use super::core::*;
+use super::lister::AliyunDriveLister;
+use super::lister::AliyunDriveParent;
+use super::reader::AliyunDriveReader;
+use super::writer::{AliyunDriveWriter, AliyunDriveWriters};
+use crate::raw::*;
+use crate::*;
+
+/// Aliyun Drive services support.
+#[derive(Default, Deserialize)]
+#[serde(default)]
+#[non_exhaustive]
+pub struct AliyunDriveConfig {
+    /// root of this backend.
+    ///
+    /// All operations will happen under this root.
+    ///
+    /// default to `/` if not set.
+    pub root: Option<String>,
+    /// client_id of this backend.
+    ///
+    /// required.
+    pub client_id: String,
+    /// client_secret of this backend.
+    ///
+    /// required.
+    pub client_secret: String,
+    /// refresh_token of this backend.
+    ///
+    /// required.
+    pub refresh_token: String,
+    /// drive_type of this backend.
+    ///
+    /// All operations will happen under this type of drive.
+    ///
+    /// default to DriveType::Default if not set.
+    pub drive_type: DriveType,
+    /// rapid_upload of this backend.

Review Comment:
   Would you like to provide more information about the `rapid_upload`?



##########
core/src/services/aliyun_drive/core.rs:
##########
@@ -0,0 +1,506 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use std::fmt::Debug;
+use std::sync::Arc;
+
+use bytes::{Buf, Bytes};
+use chrono::Utc;
+use http::header::HeaderValue;
+use http::{Method, Request};
+use serde::{Deserialize, Serialize};
+use serde_json::json;
+use tokio::sync::Mutex;
+
+use crate::raw::*;
+use crate::*;
+
+use super::error::parse_error;
+
+/// Available Aliyun Drive Type.
+#[derive(Debug, Deserialize, Default, Clone)]
+pub enum DriveType {
+    /// Use the default type of Aliyun Drive.
+    #[default]
+    Default,
+    /// Use the backup type of Aliyun Drive.
+    ///
+    /// fallback to the default type if no backup drive found.
+    Backup,
+    /// Use the resource type of Aliyun Drive.
+    ///
+    /// fallback to the default type if no resource drive found.
+    Resource,
+}
+
+pub struct AliyunDriveTokener {
+    pub drive_id: Option<String>,
+    pub access_token: Option<String>,
+    pub refresh_token: String,
+    pub expire_at: i64,
+}
+
+pub struct AliyunDriveCore {
+    pub endpoint: String,
+    pub root: String,
+    pub client_id: String,
+    pub client_secret: String,
+    pub drive_type: DriveType,
+    pub rapid_upload: bool,
+
+    pub tokener: Arc<Mutex<AliyunDriveTokener>>,
+    pub client: HttpClient,
+}
+
+impl Debug for AliyunDriveCore {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.debug_struct("AliyunDriveCore")
+            .field("root", &self.root)
+            .field("drive_type", &self.drive_type)
+            .finish_non_exhaustive()
+    }
+}
+
+impl AliyunDriveCore {
+    async fn send(&self, mut req: Request<Buffer>, token: Option<&str>) -> Result<Buffer> {
+        if req.method() == Method::POST {
+            req.headers_mut().insert(
+                "Content-Type",

Review Comment:
   Please use `http::header::CONTENT_TYPE` which takes less memory and more safe to use.



##########
core/src/services/aliyun_drive/lister.rs:
##########
@@ -0,0 +1,150 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use std::sync::Arc;
+
+use bytes::Buf;
+use chrono::Utc;
+
+use crate::raw::*;
+use crate::EntryMode;
+use crate::Error;
+use crate::ErrorKind;
+use crate::Metadata;
+use crate::Result;
+
+use self::oio::Entry;
+
+use super::core::AliyunDriveCore;
+use super::core::AliyunDriveFile;
+use super::core::AliyunDriveFileList;
+
+pub struct AliyunDriveLister {
+    core: Arc<AliyunDriveCore>,
+
+    parent: Option<AliyunDriveParent>,
+    limit: Option<usize>,
+}
+
+pub struct AliyunDriveParent {
+    pub parent_path: String,
+    pub parent_file_id: String,
+}
+
+impl AliyunDriveLister {
+    pub fn new(
+        core: Arc<AliyunDriveCore>,
+        parent: Option<AliyunDriveParent>,
+        limit: Option<usize>,
+    ) -> Self {
+        AliyunDriveLister {
+            core,
+            parent,
+            limit,
+        }
+    }
+}
+
+impl oio::PageList for AliyunDriveLister {
+    async fn next_page(&self, ctx: &mut oio::PageContext) -> Result<()> {
+        let Some(parent) = &self.parent else {
+            ctx.done = true;
+            return Ok(());
+        };
+
+        let offset = if ctx.token.is_empty() {
+            None
+        } else {
+            Some(ctx.token.clone())
+        };
+
+        let res = self
+            .core
+            .list(&parent.parent_file_id, self.limit, offset)
+            .await;
+        let res = match res {
+            Err(err) if err.kind() == ErrorKind::NotFound => {
+                ctx.done = true;
+                None
+            }
+            Err(err) => return Err(err),
+            Ok(res) => Some(res),
+        };
+
+        if let Some(res) = res {

Review Comment:
   How about returning early so we don't need to have a large block here?



##########
core/src/services/aliyun_drive/mod.rs:
##########
@@ -0,0 +1,27 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+mod backend;
+mod error;
+mod lister;
+mod reader;
+mod writer;
+pub use backend::AliyunDriveBuilder as AliyunDrive;
+pub use backend::AliyunDriveConfig;
+pub use core::DriveType as AliyunDriveType;

Review Comment:
   Please keep `AliyunDriveType` private. Only the builder and config should be accessible to users.



##########
core/src/services/aliyun_drive/core.rs:
##########
@@ -0,0 +1,506 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use std::fmt::Debug;
+use std::sync::Arc;
+
+use bytes::{Buf, Bytes};
+use chrono::Utc;
+use http::header::HeaderValue;
+use http::{Method, Request};
+use serde::{Deserialize, Serialize};
+use serde_json::json;
+use tokio::sync::Mutex;
+
+use crate::raw::*;
+use crate::*;
+
+use super::error::parse_error;
+
+/// Available Aliyun Drive Type.
+#[derive(Debug, Deserialize, Default, Clone)]
+pub enum DriveType {
+    /// Use the default type of Aliyun Drive.
+    #[default]
+    Default,
+    /// Use the backup type of Aliyun Drive.
+    ///
+    /// fallback to the default type if no backup drive found.
+    Backup,
+    /// Use the resource type of Aliyun Drive.
+    ///
+    /// fallback to the default type if no resource drive found.
+    Resource,
+}
+
+pub struct AliyunDriveTokener {
+    pub drive_id: Option<String>,
+    pub access_token: Option<String>,
+    pub refresh_token: String,
+    pub expire_at: i64,
+}
+
+pub struct AliyunDriveCore {
+    pub endpoint: String,
+    pub root: String,
+    pub client_id: String,
+    pub client_secret: String,
+    pub drive_type: DriveType,
+    pub rapid_upload: bool,
+
+    pub tokener: Arc<Mutex<AliyunDriveTokener>>,
+    pub client: HttpClient,
+}
+
+impl Debug for AliyunDriveCore {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.debug_struct("AliyunDriveCore")
+            .field("root", &self.root)
+            .field("drive_type", &self.drive_type)
+            .finish_non_exhaustive()
+    }
+}
+
+impl AliyunDriveCore {
+    async fn send(&self, mut req: Request<Buffer>, token: Option<&str>) -> Result<Buffer> {
+        if req.method() == Method::POST {
+            req.headers_mut().insert(
+                "Content-Type",
+                HeaderValue::from_static("application/json;charset=UTF-8"),
+            );
+        }
+        if let Some(token) = token {
+            req.headers_mut().insert(
+                "Authorization",
+                HeaderValue::from_str(&format!("Bearer {}", token))
+                    .expect("access token must be valid header value"),
+            );
+        }
+        let res = self.client.send(req).await?;
+        if !res.status().is_success() {
+            return Err(parse_error(res).await?);
+        }
+        Ok(res.into_body())
+    }
+
+    async fn get_access_token(&self, refresh_token: &str) -> Result<Buffer> {
+        let body = serde_json::to_vec(&json!({
+            "refresh_token": refresh_token,
+            "grant_type": "refresh_token",
+            "client_id": &self.client_id,
+            "client_secret": &self.client_secret,
+        }))
+        .map_err(new_json_serialize_error)?;
+        let req = Request::post(format!("{}/oauth/access_token", self.endpoint))
+            .body(Buffer::from(bytes::Bytes::from(body)))
+            .map_err(new_request_build_error)?;
+        self.send(req, None).await
+    }
+
+    async fn get_drive_id(&self, token: Option<&str>) -> Result<Buffer> {
+        let req = Request::post(format!("{}/adrive/v1.0/user/getDriveInfo", self.endpoint))
+            .body(Buffer::new())
+            .map_err(new_request_build_error)?;
+        self.send(req, token).await
+    }
+
+    pub async fn get_token_and_drive(&self) -> Result<(Option<String>, String)> {
+        let mut tokener = self.tokener.lock().await;
+        if tokener.expire_at < Utc::now().timestamp() || tokener.access_token.is_none() {
+            let res = self.get_access_token(&tokener.refresh_token).await?;
+            let output: RefresTokenResponse =
+                serde_json::from_reader(res.reader()).map_err(new_json_deserialize_error)?;
+            tokener.access_token = Some(output.access_token);
+            tokener.expire_at = output.expires_in + Utc::now().timestamp();
+            tokener.refresh_token = output.refresh_token;
+        }
+        let Some(drive_id) = &tokener.drive_id else {
+            let res = self.get_drive_id(tokener.access_token.as_deref()).await?;
+            let output: DriveInfoResponse =
+                serde_json::from_reader(res.reader()).map_err(new_json_deserialize_error)?;
+            let drive_id = match self.drive_type {
+                DriveType::Default => output.default_drive_id,
+                DriveType::Backup => output.backup_drive_id.unwrap_or(output.default_drive_id),
+                DriveType::Resource => output.resource_drive_id.unwrap_or(output.default_drive_id),
+            };
+            tokener.drive_id = Some(drive_id.clone());
+            return Ok((tokener.access_token.clone(), drive_id));
+        };
+        Ok((tokener.access_token.clone(), drive_id.clone()))
+    }
+
+    pub async fn get_by_path(&self, path: &str) -> Result<Buffer> {
+        let file_path = build_rooted_abs_path(&self.root, path);
+        let file_path = file_path.strip_suffix('/').unwrap_or(&file_path);
+        let req = Request::post(format!(
+            "{}/adrive/v1.0/openFile/get_by_path",
+            self.endpoint
+        ));
+        let (token, drive_id) = self.get_token_and_drive().await?;
+        let body = serde_json::to_vec(&json!({
+            "drive_id": &drive_id,
+            "file_path": &file_path,
+        }))
+        .map_err(new_json_serialize_error)?;
+        let req = req
+            .body(Buffer::from(Bytes::from(body)))
+            .map_err(new_request_build_error)?;
+        self.send(req, token.as_deref()).await
+    }
+
+    pub async fn ensure_dir_exists(&self, path: &str) -> Result<String> {
+        let file_path = build_abs_path(&self.root, path.strip_suffix('/').unwrap_or(path));
+        if file_path == "/" {
+            return Ok("root".to_string());
+        }
+        let file_path = file_path.strip_suffix('/').unwrap_or(&file_path);
+        let paths = file_path.split('/').collect::<Vec<&str>>();
+        let mut parent: Option<String> = None;
+        for path in paths {
+            let res = self
+                .create(
+                    parent.as_deref(),
+                    path,
+                    CreateType::Folder,
+                    CheckNameMode::Refuse,
+                )
+                .await?;
+            let output: CreateResponse =
+                serde_json::from_reader(res.reader()).map_err(new_json_deserialize_error)?;
+            parent = Some(output.file_id);
+        }
+        Ok(parent.expect("ensure_dir_exists must succeed"))
+    }
+
+    pub async fn create_with_rapid_upload(
+        &self,
+        parent_file_id: Option<&str>,
+        name: &str,
+        typ: CreateType,
+        check_name_mode: CheckNameMode,
+        size: Option<u64>,
+        rapid_upload: Option<RapidUpload>,
+    ) -> Result<Buffer> {
+        let mut content_hash = None;
+        let mut proof_code = None;
+        let mut pre_hash = None;
+        if let Some(rapid_upload) = rapid_upload {
+            content_hash = rapid_upload.content_hash;
+            proof_code = rapid_upload.proof_code;
+            pre_hash = rapid_upload.pre_hash;
+        }
+
+        let (token, drive_id) = self.get_token_and_drive().await?;
+        let body = serde_json::to_vec(&json!({
+            "drive_id": &drive_id,
+            "parent_file_id": parent_file_id.unwrap_or("root"),
+            "name": name,
+            "type": typ,
+            "check_name_mode": check_name_mode,
+            "size": size,
+            "pre_hash": pre_hash,
+            "content_hash": content_hash,
+            "content_hash_name": content_hash.is_some().then_some("sha1"),
+            "proof_code": proof_code,
+            "proof_version": proof_code.is_some().then_some("v1"),
+        }))
+        .map_err(new_json_serialize_error)?;
+        let req = Request::post(format!("{}/adrive/v1.0/openFile/create", self.endpoint))
+            .body(Buffer::from(Bytes::from(body)))
+            .map_err(new_request_build_error)?;
+        self.send(req, token.as_deref()).await
+    }
+
+    pub async fn create(
+        &self,
+        parent_file_id: Option<&str>,
+        name: &str,
+        typ: CreateType,
+        check_name_mode: CheckNameMode,
+    ) -> Result<Buffer> {
+        self.create_with_rapid_upload(parent_file_id, name, typ, check_name_mode, None, None)
+            .await
+    }
+
+    pub async fn get_download_url(&self, file_id: &str) -> Result<String> {
+        let (token, drive_id) = self.get_token_and_drive().await?;
+        let body = serde_json::to_vec(&json!({
+            "drive_id": &drive_id,
+            "file_id": file_id,
+        }))
+        .map_err(new_json_serialize_error)?;
+        let req = Request::post(format!(
+            "{}/adrive/v1.0/openFile/getDownloadUrl",
+            self.endpoint
+        ))
+        .body(Buffer::from(Bytes::from(body)))
+        .map_err(new_request_build_error)?;
+        let res = self.send(req, token.as_deref()).await?;
+        let output: GetDownloadUrlResponse =
+            serde_json::from_reader(res.reader()).map_err(new_json_serialize_error)?;
+        Ok(output.url)
+    }
+
+    pub async fn move_path(&self, file_id: &str, to_parent_file_id: &str) -> Result<()> {
+        let (token, drive_id) = self.get_token_and_drive().await?;
+        let body = serde_json::to_vec(&json!({
+            "drive_id": &drive_id,
+            "file_id": file_id,
+            "to_parent_file_id": to_parent_file_id,
+            "check_name_mode": CheckNameMode::AutoRename,
+        }))
+        .map_err(new_json_serialize_error)?;
+        let req = Request::post(format!("{}/adrive/v1.0/openFile/move", self.endpoint))
+            .body(Buffer::from(Bytes::from(body)))
+            .map_err(new_request_build_error)?;
+        self.send(req, token.as_deref()).await?;
+        Ok(())
+    }
+
+    pub async fn update_path(&self, file_id: &str, name: &str) -> Result<()> {
+        let (token, drive_id) = self.get_token_and_drive().await?;
+        let body = serde_json::to_vec(&json!({
+            "drive_id": &drive_id,
+            "file_id": file_id,
+            "name": name,
+            "check_name_mode": CheckNameMode::Refuse,
+        }))
+        .map_err(new_json_serialize_error)?;
+        let req = Request::post(format!("{}/adrive/v1.0/openFile/update", self.endpoint))
+            .body(Buffer::from(Bytes::from(body)))

Review Comment:
   Buffer can be built from `Vec<u8>` directly, so we don't need to build `Bytes` first.



##########
core/src/services/mod.rs:
##########
@@ -19,6 +19,15 @@
 //!
 //! More ongoing services support is tracked at [opendal#5](https://github.com/apache/opendal/issues/5). Please feel free to submit issues if there are services not covered.
 
+#[cfg(feature = "services-aliyun-drive")]
+mod aliyun_drive;
+#[cfg(feature = "services-aliyun-drive")]
+pub use aliyun_drive::AliyunDrive;
+#[cfg(feature = "services-aliyun-drive")]
+pub use aliyun_drive::AliyunDriveConfig;
+#[cfg(feature = "services-aliyun-drive")]
+pub use aliyun_drive::AliyunDriveType;

Review Comment:
   the same.



##########
core/src/services/aliyun_drive/core.rs:
##########
@@ -0,0 +1,506 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use std::fmt::Debug;
+use std::sync::Arc;
+
+use bytes::{Buf, Bytes};
+use chrono::Utc;
+use http::header::HeaderValue;
+use http::{Method, Request};
+use serde::{Deserialize, Serialize};
+use serde_json::json;
+use tokio::sync::Mutex;
+
+use crate::raw::*;
+use crate::*;
+
+use super::error::parse_error;
+
+/// Available Aliyun Drive Type.
+#[derive(Debug, Deserialize, Default, Clone)]
+pub enum DriveType {
+    /// Use the default type of Aliyun Drive.
+    #[default]
+    Default,
+    /// Use the backup type of Aliyun Drive.
+    ///
+    /// fallback to the default type if no backup drive found.
+    Backup,
+    /// Use the resource type of Aliyun Drive.
+    ///
+    /// fallback to the default type if no resource drive found.
+    Resource,
+}
+
+pub struct AliyunDriveTokener {
+    pub drive_id: Option<String>,
+    pub access_token: Option<String>,
+    pub refresh_token: String,
+    pub expire_at: i64,
+}
+
+pub struct AliyunDriveCore {
+    pub endpoint: String,
+    pub root: String,
+    pub client_id: String,
+    pub client_secret: String,
+    pub drive_type: DriveType,
+    pub rapid_upload: bool,
+
+    pub tokener: Arc<Mutex<AliyunDriveTokener>>,
+    pub client: HttpClient,
+}
+
+impl Debug for AliyunDriveCore {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.debug_struct("AliyunDriveCore")
+            .field("root", &self.root)
+            .field("drive_type", &self.drive_type)
+            .finish_non_exhaustive()
+    }
+}
+
+impl AliyunDriveCore {
+    async fn send(&self, mut req: Request<Buffer>, token: Option<&str>) -> Result<Buffer> {
+        if req.method() == Method::POST {
+            req.headers_mut().insert(
+                "Content-Type",
+                HeaderValue::from_static("application/json;charset=UTF-8"),
+            );
+        }
+        if let Some(token) = token {
+            req.headers_mut().insert(
+                "Authorization",

Review Comment:
   the same.



##########
core/src/services/aliyun_drive/core.rs:
##########
@@ -0,0 +1,506 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use std::fmt::Debug;
+use std::sync::Arc;
+
+use bytes::{Buf, Bytes};
+use chrono::Utc;
+use http::header::HeaderValue;
+use http::{Method, Request};
+use serde::{Deserialize, Serialize};
+use serde_json::json;
+use tokio::sync::Mutex;
+
+use crate::raw::*;
+use crate::*;
+
+use super::error::parse_error;
+
+/// Available Aliyun Drive Type.
+#[derive(Debug, Deserialize, Default, Clone)]
+pub enum DriveType {
+    /// Use the default type of Aliyun Drive.
+    #[default]
+    Default,
+    /// Use the backup type of Aliyun Drive.
+    ///
+    /// fallback to the default type if no backup drive found.
+    Backup,
+    /// Use the resource type of Aliyun Drive.
+    ///
+    /// fallback to the default type if no resource drive found.
+    Resource,
+}
+
+pub struct AliyunDriveTokener {
+    pub drive_id: Option<String>,
+    pub access_token: Option<String>,
+    pub refresh_token: String,
+    pub expire_at: i64,
+}
+
+pub struct AliyunDriveCore {
+    pub endpoint: String,
+    pub root: String,
+    pub client_id: String,
+    pub client_secret: String,
+    pub drive_type: DriveType,
+    pub rapid_upload: bool,
+
+    pub tokener: Arc<Mutex<AliyunDriveTokener>>,
+    pub client: HttpClient,
+}
+
+impl Debug for AliyunDriveCore {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.debug_struct("AliyunDriveCore")
+            .field("root", &self.root)
+            .field("drive_type", &self.drive_type)
+            .finish_non_exhaustive()
+    }
+}
+
+impl AliyunDriveCore {
+    async fn send(&self, mut req: Request<Buffer>, token: Option<&str>) -> Result<Buffer> {
+        if req.method() == Method::POST {
+            req.headers_mut().insert(
+                "Content-Type",
+                HeaderValue::from_static("application/json;charset=UTF-8"),
+            );
+        }
+        if let Some(token) = token {
+            req.headers_mut().insert(
+                "Authorization",
+                HeaderValue::from_str(&format!("Bearer {}", token))
+                    .expect("access token must be valid header value"),
+            );
+        }
+        let res = self.client.send(req).await?;
+        if !res.status().is_success() {
+            return Err(parse_error(res).await?);
+        }
+        Ok(res.into_body())
+    }
+
+    async fn get_access_token(&self, refresh_token: &str) -> Result<Buffer> {
+        let body = serde_json::to_vec(&json!({
+            "refresh_token": refresh_token,
+            "grant_type": "refresh_token",
+            "client_id": &self.client_id,
+            "client_secret": &self.client_secret,
+        }))
+        .map_err(new_json_serialize_error)?;
+        let req = Request::post(format!("{}/oauth/access_token", self.endpoint))
+            .body(Buffer::from(bytes::Bytes::from(body)))
+            .map_err(new_request_build_error)?;
+        self.send(req, None).await
+    }
+
+    async fn get_drive_id(&self, token: Option<&str>) -> Result<Buffer> {
+        let req = Request::post(format!("{}/adrive/v1.0/user/getDriveInfo", self.endpoint))
+            .body(Buffer::new())
+            .map_err(new_request_build_error)?;
+        self.send(req, token).await
+    }
+
+    pub async fn get_token_and_drive(&self) -> Result<(Option<String>, String)> {
+        let mut tokener = self.tokener.lock().await;
+        if tokener.expire_at < Utc::now().timestamp() || tokener.access_token.is_none() {
+            let res = self.get_access_token(&tokener.refresh_token).await?;
+            let output: RefresTokenResponse =
+                serde_json::from_reader(res.reader()).map_err(new_json_deserialize_error)?;
+            tokener.access_token = Some(output.access_token);
+            tokener.expire_at = output.expires_in + Utc::now().timestamp();
+            tokener.refresh_token = output.refresh_token;
+        }
+        let Some(drive_id) = &tokener.drive_id else {
+            let res = self.get_drive_id(tokener.access_token.as_deref()).await?;
+            let output: DriveInfoResponse =
+                serde_json::from_reader(res.reader()).map_err(new_json_deserialize_error)?;
+            let drive_id = match self.drive_type {
+                DriveType::Default => output.default_drive_id,
+                DriveType::Backup => output.backup_drive_id.unwrap_or(output.default_drive_id),
+                DriveType::Resource => output.resource_drive_id.unwrap_or(output.default_drive_id),
+            };
+            tokener.drive_id = Some(drive_id.clone());
+            return Ok((tokener.access_token.clone(), drive_id));
+        };
+        Ok((tokener.access_token.clone(), drive_id.clone()))
+    }
+
+    pub async fn get_by_path(&self, path: &str) -> Result<Buffer> {
+        let file_path = build_rooted_abs_path(&self.root, path);
+        let file_path = file_path.strip_suffix('/').unwrap_or(&file_path);
+        let req = Request::post(format!(
+            "{}/adrive/v1.0/openFile/get_by_path",
+            self.endpoint
+        ));
+        let (token, drive_id) = self.get_token_and_drive().await?;
+        let body = serde_json::to_vec(&json!({
+            "drive_id": &drive_id,
+            "file_path": &file_path,
+        }))
+        .map_err(new_json_serialize_error)?;
+        let req = req
+            .body(Buffer::from(Bytes::from(body)))
+            .map_err(new_request_build_error)?;
+        self.send(req, token.as_deref()).await
+    }
+
+    pub async fn ensure_dir_exists(&self, path: &str) -> Result<String> {
+        let file_path = build_abs_path(&self.root, path.strip_suffix('/').unwrap_or(path));
+        if file_path == "/" {
+            return Ok("root".to_string());
+        }
+        let file_path = file_path.strip_suffix('/').unwrap_or(&file_path);
+        let paths = file_path.split('/').collect::<Vec<&str>>();
+        let mut parent: Option<String> = None;
+        for path in paths {
+            let res = self
+                .create(
+                    parent.as_deref(),
+                    path,
+                    CreateType::Folder,
+                    CheckNameMode::Refuse,
+                )
+                .await?;
+            let output: CreateResponse =
+                serde_json::from_reader(res.reader()).map_err(new_json_deserialize_error)?;
+            parent = Some(output.file_id);
+        }
+        Ok(parent.expect("ensure_dir_exists must succeed"))
+    }
+
+    pub async fn create_with_rapid_upload(
+        &self,
+        parent_file_id: Option<&str>,
+        name: &str,
+        typ: CreateType,
+        check_name_mode: CheckNameMode,
+        size: Option<u64>,
+        rapid_upload: Option<RapidUpload>,
+    ) -> Result<Buffer> {
+        let mut content_hash = None;
+        let mut proof_code = None;
+        let mut pre_hash = None;
+        if let Some(rapid_upload) = rapid_upload {
+            content_hash = rapid_upload.content_hash;
+            proof_code = rapid_upload.proof_code;
+            pre_hash = rapid_upload.pre_hash;
+        }
+
+        let (token, drive_id) = self.get_token_and_drive().await?;
+        let body = serde_json::to_vec(&json!({
+            "drive_id": &drive_id,
+            "parent_file_id": parent_file_id.unwrap_or("root"),
+            "name": name,
+            "type": typ,
+            "check_name_mode": check_name_mode,
+            "size": size,
+            "pre_hash": pre_hash,
+            "content_hash": content_hash,
+            "content_hash_name": content_hash.is_some().then_some("sha1"),
+            "proof_code": proof_code,
+            "proof_version": proof_code.is_some().then_some("v1"),
+        }))
+        .map_err(new_json_serialize_error)?;
+        let req = Request::post(format!("{}/adrive/v1.0/openFile/create", self.endpoint))
+            .body(Buffer::from(Bytes::from(body)))
+            .map_err(new_request_build_error)?;
+        self.send(req, token.as_deref()).await
+    }
+
+    pub async fn create(
+        &self,
+        parent_file_id: Option<&str>,
+        name: &str,
+        typ: CreateType,
+        check_name_mode: CheckNameMode,
+    ) -> Result<Buffer> {
+        self.create_with_rapid_upload(parent_file_id, name, typ, check_name_mode, None, None)
+            .await
+    }
+
+    pub async fn get_download_url(&self, file_id: &str) -> Result<String> {
+        let (token, drive_id) = self.get_token_and_drive().await?;
+        let body = serde_json::to_vec(&json!({
+            "drive_id": &drive_id,
+            "file_id": file_id,
+        }))
+        .map_err(new_json_serialize_error)?;
+        let req = Request::post(format!(
+            "{}/adrive/v1.0/openFile/getDownloadUrl",
+            self.endpoint
+        ))
+        .body(Buffer::from(Bytes::from(body)))
+        .map_err(new_request_build_error)?;
+        let res = self.send(req, token.as_deref()).await?;
+        let output: GetDownloadUrlResponse =
+            serde_json::from_reader(res.reader()).map_err(new_json_serialize_error)?;
+        Ok(output.url)
+    }
+
+    pub async fn move_path(&self, file_id: &str, to_parent_file_id: &str) -> Result<()> {
+        let (token, drive_id) = self.get_token_and_drive().await?;
+        let body = serde_json::to_vec(&json!({
+            "drive_id": &drive_id,
+            "file_id": file_id,
+            "to_parent_file_id": to_parent_file_id,
+            "check_name_mode": CheckNameMode::AutoRename,
+        }))
+        .map_err(new_json_serialize_error)?;
+        let req = Request::post(format!("{}/adrive/v1.0/openFile/move", self.endpoint))
+            .body(Buffer::from(Bytes::from(body)))
+            .map_err(new_request_build_error)?;
+        self.send(req, token.as_deref()).await?;
+        Ok(())
+    }
+
+    pub async fn update_path(&self, file_id: &str, name: &str) -> Result<()> {
+        let (token, drive_id) = self.get_token_and_drive().await?;
+        let body = serde_json::to_vec(&json!({

Review Comment:
   How about using a struct instead of building JSON with `json!()`? I think it will be easier to use and less prone to errors.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@opendal.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] feat(services/aliyun-drive): support AliyunDrive [opendal]

["yuchanns (via GitHub)" <gi...@apache.org>]

yuchanns commented on PR #4585:
URL: https://github.com/apache/opendal/pull/4585#issuecomment-2117576840

   > it's not a requirement to implement MultipartWriter
   
   Well. that's fine. I will raise another PR to do it.


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@opendal.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org