You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@opendal.apache.org by "Xuanwo (via GitHub)" <gi...@apache.org> on 2023/10/19 15:27:01 UTC
Re: [PR] feat(service/cloudflare_kv): support cloudflare KV [incubator-opendal]
Xuanwo commented on code in PR #3348:
URL: https://github.com/apache/incubator-opendal/pull/3348#discussion_r1365710024
##########
core/src/services/cloudflare_kv/backend.rs:
##########
@@ -0,0 +1,444 @@
+// 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.
+#![allow(dead_code)]
+use std::collections::HashMap;
+use std::fmt::Debug;
+use std::fmt::Formatter;
+
+use async_trait::async_trait;
+use http::header;
+use http::Request;
+use http::Response;
+use http::StatusCode;
+use serde::Deserialize;
+
+use super::error::parse_error;
+use crate::raw::adapters::kv;
+use crate::raw::*;
+use crate::ErrorKind;
+use crate::*;
+
+#[doc = include_str!("docs.md")]
+#[derive(Default)]
+pub struct CloudflareKvBuilder {
+ /// The token used to authenticate with CloudFlare.
+ token: Option<String>,
+ /// The account ID used to authenticate with CloudFlare. Used as URI path parameter.
+ account_id: Option<String>,
+ /// The namespace ID. Used as URI path parameter.
+ namespace_id: Option<String>,
+
+ /// The HTTP client used to communicate with CloudFlare.
+ http_client: Option<HttpClient>,
+ /// Root within this backend.
+ root: Option<String>,
+}
+
+impl Debug for CloudflareKvBuilder {
+ fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+ f.debug_struct("CloudFlareKvBuilder")
+ .field("account_id", &self.account_id)
+ .field("namespace_id", &self.namespace_id)
+ .field("root", &self.root)
+ .finish()
+ }
+}
+
+impl CloudflareKvBuilder {
+ /// Set the token used to authenticate with CloudFlare.
+ pub fn token(&mut self, token: &str) -> &mut Self {
+ if !token.is_empty() {
+ self.token = Some(token.to_string())
+ }
+ self
+ }
+
+ /// Set the account ID used to authenticate with CloudFlare.
+ pub fn account_id(&mut self, account_id: &str) -> &mut Self {
+ if !account_id.is_empty() {
+ self.account_id = Some(account_id.to_string())
+ }
+ self
+ }
+
+ /// Set the namespace ID.
+ pub fn namespace_id(&mut self, namespace_id: &str) -> &mut Self {
+ if !namespace_id.is_empty() {
+ self.namespace_id = Some(namespace_id.to_string())
+ }
+ self
+ }
+
+ /// Set the root within this backend.
+ pub fn root(&mut self, root: &str) -> &mut Self {
+ if !root.is_empty() {
+ self.root = Some(root.to_string())
+ }
+ self
+ }
+}
+
+impl Builder for CloudflareKvBuilder {
+ const SCHEME: Scheme = Scheme::CloudflareKv;
+
+ type Accessor = CloudflareKvBackend;
+
+ fn from_map(map: HashMap<String, String>) -> Self {
+ let mut builder = Self::default();
+ map.get("token").map(|v| builder.token(v));
+ map.get("account_id").map(|v| builder.account_id(v));
+ map.get("namespace_id").map(|v| builder.namespace_id(v));
+ map.get("root").map(|v| builder.root(v));
+ builder
+ }
+
+ fn build(&mut self) -> Result<Self::Accessor> {
+ let mut authorization = None;
+ if let Some(token) = &self.token {
+ authorization = Some(format_authorization_by_bearer(token)?)
+ }
+
+ let Some(account_id) = self.account_id.clone() else {
+ return Err(Error::new(
+ ErrorKind::ConfigInvalid,
+ "account_id is required",
+ ));
+ };
+
+ let Some(namespace_id) = self.namespace_id.clone() else {
+ return Err(Error::new(
+ ErrorKind::ConfigInvalid,
+ "namespace_id is required",
+ ));
+ };
+
+ 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::CloudflareKv)
+ })?
+ };
+
+ let root = normalize_root(
+ self.root
+ .clone()
+ .unwrap_or_else(|| "/".to_string())
+ .as_str(),
+ );
+
+ Ok(kv::Backend::new(Adapter {
+ authorization,
+ account_id,
+ namespace_id,
+ client,
+ })
+ .with_root(&root))
+ }
+}
+
+pub type CloudflareKvBackend = kv::Backend<Adapter>;
+
+#[derive(Clone)]
+pub struct Adapter {
+ authorization: Option<String>,
+ account_id: String,
+ namespace_id: String,
+ client: HttpClient,
+}
+
+impl Debug for Adapter {
+ fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+ f.debug_struct("Adapter")
+ .field("account_id", &self.account_id)
+ .field("namespace_id", &self.namespace_id)
+ .finish()
+ }
+}
+
+impl Adapter {
+ fn load_token<T>(&self, mut req: Request<T>) -> Result<Request<T>> {
Review Comment:
How about using `sign()` to align the API naming style?
##########
core/src/services/cloudflare_kv/backend.rs:
##########
@@ -0,0 +1,444 @@
+// 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.
+#![allow(dead_code)]
+use std::collections::HashMap;
+use std::fmt::Debug;
+use std::fmt::Formatter;
+
+use async_trait::async_trait;
+use http::header;
+use http::Request;
+use http::Response;
+use http::StatusCode;
+use serde::Deserialize;
+
+use super::error::parse_error;
+use crate::raw::adapters::kv;
+use crate::raw::*;
+use crate::ErrorKind;
+use crate::*;
+
+#[doc = include_str!("docs.md")]
+#[derive(Default)]
+pub struct CloudflareKvBuilder {
+ /// The token used to authenticate with CloudFlare.
+ token: Option<String>,
+ /// The account ID used to authenticate with CloudFlare. Used as URI path parameter.
+ account_id: Option<String>,
+ /// The namespace ID. Used as URI path parameter.
+ namespace_id: Option<String>,
+
+ /// The HTTP client used to communicate with CloudFlare.
+ http_client: Option<HttpClient>,
+ /// Root within this backend.
+ root: Option<String>,
+}
+
+impl Debug for CloudflareKvBuilder {
+ fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+ f.debug_struct("CloudFlareKvBuilder")
+ .field("account_id", &self.account_id)
+ .field("namespace_id", &self.namespace_id)
+ .field("root", &self.root)
+ .finish()
+ }
+}
+
+impl CloudflareKvBuilder {
+ /// Set the token used to authenticate with CloudFlare.
+ pub fn token(&mut self, token: &str) -> &mut Self {
+ if !token.is_empty() {
+ self.token = Some(token.to_string())
+ }
+ self
+ }
+
+ /// Set the account ID used to authenticate with CloudFlare.
+ pub fn account_id(&mut self, account_id: &str) -> &mut Self {
+ if !account_id.is_empty() {
+ self.account_id = Some(account_id.to_string())
+ }
+ self
+ }
+
+ /// Set the namespace ID.
+ pub fn namespace_id(&mut self, namespace_id: &str) -> &mut Self {
+ if !namespace_id.is_empty() {
+ self.namespace_id = Some(namespace_id.to_string())
+ }
+ self
+ }
+
+ /// Set the root within this backend.
+ pub fn root(&mut self, root: &str) -> &mut Self {
+ if !root.is_empty() {
+ self.root = Some(root.to_string())
+ }
+ self
+ }
+}
+
+impl Builder for CloudflareKvBuilder {
+ const SCHEME: Scheme = Scheme::CloudflareKv;
+
+ type Accessor = CloudflareKvBackend;
+
+ fn from_map(map: HashMap<String, String>) -> Self {
+ let mut builder = Self::default();
+ map.get("token").map(|v| builder.token(v));
+ map.get("account_id").map(|v| builder.account_id(v));
+ map.get("namespace_id").map(|v| builder.namespace_id(v));
+ map.get("root").map(|v| builder.root(v));
+ builder
+ }
+
+ fn build(&mut self) -> Result<Self::Accessor> {
+ let mut authorization = None;
+ if let Some(token) = &self.token {
+ authorization = Some(format_authorization_by_bearer(token)?)
+ }
+
+ let Some(account_id) = self.account_id.clone() else {
Review Comment:
Cool!
##########
core/src/services/cloudflare_kv/backend.rs:
##########
@@ -0,0 +1,444 @@
+// 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.
+#![allow(dead_code)]
Review Comment:
We don't permit `dead_code` unless there's a compelling reason. Is there one?
##########
core/src/services/cloudflare_kv/backend.rs:
##########
@@ -0,0 +1,444 @@
+// 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.
+#![allow(dead_code)]
+use std::collections::HashMap;
+use std::fmt::Debug;
+use std::fmt::Formatter;
+
+use async_trait::async_trait;
+use http::header;
+use http::Request;
+use http::Response;
+use http::StatusCode;
+use serde::Deserialize;
+
+use super::error::parse_error;
+use crate::raw::adapters::kv;
+use crate::raw::*;
+use crate::ErrorKind;
+use crate::*;
+
+#[doc = include_str!("docs.md")]
+#[derive(Default)]
+pub struct CloudflareKvBuilder {
+ /// The token used to authenticate with CloudFlare.
+ token: Option<String>,
+ /// The account ID used to authenticate with CloudFlare. Used as URI path parameter.
+ account_id: Option<String>,
+ /// The namespace ID. Used as URI path parameter.
+ namespace_id: Option<String>,
+
+ /// The HTTP client used to communicate with CloudFlare.
+ http_client: Option<HttpClient>,
+ /// Root within this backend.
+ root: Option<String>,
+}
+
+impl Debug for CloudflareKvBuilder {
+ fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+ f.debug_struct("CloudFlareKvBuilder")
+ .field("account_id", &self.account_id)
+ .field("namespace_id", &self.namespace_id)
+ .field("root", &self.root)
+ .finish()
+ }
+}
+
+impl CloudflareKvBuilder {
+ /// Set the token used to authenticate with CloudFlare.
+ pub fn token(&mut self, token: &str) -> &mut Self {
+ if !token.is_empty() {
+ self.token = Some(token.to_string())
+ }
+ self
+ }
+
+ /// Set the account ID used to authenticate with CloudFlare.
+ pub fn account_id(&mut self, account_id: &str) -> &mut Self {
+ if !account_id.is_empty() {
+ self.account_id = Some(account_id.to_string())
+ }
+ self
+ }
+
+ /// Set the namespace ID.
+ pub fn namespace_id(&mut self, namespace_id: &str) -> &mut Self {
+ if !namespace_id.is_empty() {
+ self.namespace_id = Some(namespace_id.to_string())
+ }
+ self
+ }
+
+ /// Set the root within this backend.
+ pub fn root(&mut self, root: &str) -> &mut Self {
+ if !root.is_empty() {
+ self.root = Some(root.to_string())
+ }
+ self
+ }
+}
+
+impl Builder for CloudflareKvBuilder {
+ const SCHEME: Scheme = Scheme::CloudflareKv;
+
+ type Accessor = CloudflareKvBackend;
+
+ fn from_map(map: HashMap<String, String>) -> Self {
+ let mut builder = Self::default();
+ map.get("token").map(|v| builder.token(v));
+ map.get("account_id").map(|v| builder.account_id(v));
+ map.get("namespace_id").map(|v| builder.namespace_id(v));
+ map.get("root").map(|v| builder.root(v));
+ builder
+ }
+
+ fn build(&mut self) -> Result<Self::Accessor> {
+ let mut authorization = None;
+ if let Some(token) = &self.token {
+ authorization = Some(format_authorization_by_bearer(token)?)
+ }
+
+ let Some(account_id) = self.account_id.clone() else {
+ return Err(Error::new(
+ ErrorKind::ConfigInvalid,
+ "account_id is required",
+ ));
+ };
+
+ let Some(namespace_id) = self.namespace_id.clone() else {
+ return Err(Error::new(
+ ErrorKind::ConfigInvalid,
+ "namespace_id is required",
+ ));
+ };
+
+ 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::CloudflareKv)
+ })?
+ };
+
+ let root = normalize_root(
+ self.root
+ .clone()
+ .unwrap_or_else(|| "/".to_string())
+ .as_str(),
+ );
+
+ Ok(kv::Backend::new(Adapter {
+ authorization,
+ account_id,
+ namespace_id,
+ client,
+ })
+ .with_root(&root))
+ }
+}
+
+pub type CloudflareKvBackend = kv::Backend<Adapter>;
+
+#[derive(Clone)]
+pub struct Adapter {
+ authorization: Option<String>,
Review Comment:
Can Cloudflare KV be accessed anonymously? If it doesn't, we could use `String` in place of `Option<String>`.
##########
core/src/services/cloudflare_kv/backend.rs:
##########
@@ -0,0 +1,444 @@
+// 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.
+#![allow(dead_code)]
+use std::collections::HashMap;
+use std::fmt::Debug;
+use std::fmt::Formatter;
+
+use async_trait::async_trait;
+use http::header;
+use http::Request;
+use http::Response;
+use http::StatusCode;
+use serde::Deserialize;
+
+use super::error::parse_error;
+use crate::raw::adapters::kv;
+use crate::raw::*;
+use crate::ErrorKind;
+use crate::*;
+
+#[doc = include_str!("docs.md")]
+#[derive(Default)]
+pub struct CloudflareKvBuilder {
+ /// The token used to authenticate with CloudFlare.
+ token: Option<String>,
+ /// The account ID used to authenticate with CloudFlare. Used as URI path parameter.
+ account_id: Option<String>,
+ /// The namespace ID. Used as URI path parameter.
+ namespace_id: Option<String>,
+
+ /// The HTTP client used to communicate with CloudFlare.
+ http_client: Option<HttpClient>,
+ /// Root within this backend.
+ root: Option<String>,
+}
+
+impl Debug for CloudflareKvBuilder {
+ fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+ f.debug_struct("CloudFlareKvBuilder")
+ .field("account_id", &self.account_id)
+ .field("namespace_id", &self.namespace_id)
+ .field("root", &self.root)
+ .finish()
+ }
+}
+
+impl CloudflareKvBuilder {
+ /// Set the token used to authenticate with CloudFlare.
+ pub fn token(&mut self, token: &str) -> &mut Self {
+ if !token.is_empty() {
+ self.token = Some(token.to_string())
+ }
+ self
+ }
+
+ /// Set the account ID used to authenticate with CloudFlare.
+ pub fn account_id(&mut self, account_id: &str) -> &mut Self {
+ if !account_id.is_empty() {
+ self.account_id = Some(account_id.to_string())
+ }
+ self
+ }
+
+ /// Set the namespace ID.
+ pub fn namespace_id(&mut self, namespace_id: &str) -> &mut Self {
+ if !namespace_id.is_empty() {
+ self.namespace_id = Some(namespace_id.to_string())
+ }
+ self
+ }
+
+ /// Set the root within this backend.
+ pub fn root(&mut self, root: &str) -> &mut Self {
+ if !root.is_empty() {
+ self.root = Some(root.to_string())
+ }
+ self
+ }
+}
+
+impl Builder for CloudflareKvBuilder {
+ const SCHEME: Scheme = Scheme::CloudflareKv;
+
+ type Accessor = CloudflareKvBackend;
+
+ fn from_map(map: HashMap<String, String>) -> Self {
+ let mut builder = Self::default();
+ map.get("token").map(|v| builder.token(v));
+ map.get("account_id").map(|v| builder.account_id(v));
+ map.get("namespace_id").map(|v| builder.namespace_id(v));
+ map.get("root").map(|v| builder.root(v));
+ builder
+ }
+
+ fn build(&mut self) -> Result<Self::Accessor> {
+ let mut authorization = None;
+ if let Some(token) = &self.token {
+ authorization = Some(format_authorization_by_bearer(token)?)
+ }
+
+ let Some(account_id) = self.account_id.clone() else {
+ return Err(Error::new(
+ ErrorKind::ConfigInvalid,
+ "account_id is required",
+ ));
+ };
+
+ let Some(namespace_id) = self.namespace_id.clone() else {
+ return Err(Error::new(
+ ErrorKind::ConfigInvalid,
+ "namespace_id is required",
+ ));
+ };
+
+ 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::CloudflareKv)
+ })?
+ };
+
+ let root = normalize_root(
+ self.root
+ .clone()
+ .unwrap_or_else(|| "/".to_string())
+ .as_str(),
+ );
+
+ Ok(kv::Backend::new(Adapter {
+ authorization,
+ account_id,
+ namespace_id,
+ client,
+ })
+ .with_root(&root))
+ }
+}
+
+pub type CloudflareKvBackend = kv::Backend<Adapter>;
+
+#[derive(Clone)]
+pub struct Adapter {
+ authorization: Option<String>,
+ account_id: String,
+ namespace_id: String,
+ client: HttpClient,
+}
+
+impl Debug for Adapter {
+ fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+ f.debug_struct("Adapter")
+ .field("account_id", &self.account_id)
+ .field("namespace_id", &self.namespace_id)
+ .finish()
+ }
+}
+
+impl Adapter {
+ fn load_token<T>(&self, mut req: Request<T>) -> Result<Request<T>> {
+ if let Some(auth) = &self.authorization {
+ req.headers_mut()
+ .insert(header::AUTHORIZATION, auth.parse().unwrap());
+ }
+ Ok(req)
+ }
+
+ #[inline]
+ pub async fn send(&self, req: Request<AsyncBody>) -> Result<Response<IncomingAsyncBody>> {
+ self.client.send(req).await
+ }
+}
+
+#[async_trait]
+impl kv::Adapter for Adapter {
+ fn metadata(&self) -> kv::Metadata {
+ kv::Metadata::new(
+ Scheme::CloudflareKv,
+ &self.namespace_id,
+ Capability {
+ read: true,
+ write: true,
+ list: true,
+
+ ..Default::default()
+ },
+ )
+ }
+
+ async fn get(&self, path: &str) -> Result<Option<Vec<u8>>> {
+ let url = format!(
+ r"https://api.cloudflare.com/client/v4/accounts/{}/storage/kv/namespaces/{}",
+ self.account_id, self.namespace_id
+ );
+ let url = format!("{}/values/{}", url, path);
+ let mut req = Request::get(&url);
+ req = req.header(header::CONTENT_TYPE, "application/json");
+ let mut req = req
+ .body(AsyncBody::Empty)
+ .map_err(new_request_build_error)?;
+ req = self.load_token(req)?;
+ let resp = self.client.send(req).await?;
+ let status = resp.status();
+ match status {
+ StatusCode::OK => {
+ let body = resp.into_body().bytes().await?;
+ Ok(Some(body.into()))
+ }
+ _ => Err(parse_error(resp).await?),
+ }
+ }
+
+ async fn set(&self, path: &str, value: &[u8]) -> Result<()> {
+ let url = format!(
+ r"https://api.cloudflare.com/client/v4/accounts/{}/storage/kv/namespaces/{}",
+ self.account_id, self.namespace_id
+ );
+ let url = format!("{}/values/{}", url, path);
+ let mut req = Request::put(&url);
+ req = req.header(header::CONTENT_TYPE, "multipart/form-data");
Review Comment:
Please don't set content_type by hand. `Multipart::apply` will handle this.
##########
core/src/services/cloudflare_kv/backend.rs:
##########
@@ -0,0 +1,444 @@
+// 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.
+#![allow(dead_code)]
+use std::collections::HashMap;
+use std::fmt::Debug;
+use std::fmt::Formatter;
+
+use async_trait::async_trait;
+use http::header;
+use http::Request;
+use http::Response;
+use http::StatusCode;
+use serde::Deserialize;
+
+use super::error::parse_error;
+use crate::raw::adapters::kv;
+use crate::raw::*;
+use crate::ErrorKind;
+use crate::*;
+
+#[doc = include_str!("docs.md")]
+#[derive(Default)]
+pub struct CloudflareKvBuilder {
+ /// The token used to authenticate with CloudFlare.
+ token: Option<String>,
+ /// The account ID used to authenticate with CloudFlare. Used as URI path parameter.
+ account_id: Option<String>,
+ /// The namespace ID. Used as URI path parameter.
+ namespace_id: Option<String>,
+
+ /// The HTTP client used to communicate with CloudFlare.
+ http_client: Option<HttpClient>,
+ /// Root within this backend.
+ root: Option<String>,
+}
+
+impl Debug for CloudflareKvBuilder {
+ fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+ f.debug_struct("CloudFlareKvBuilder")
+ .field("account_id", &self.account_id)
+ .field("namespace_id", &self.namespace_id)
+ .field("root", &self.root)
+ .finish()
+ }
+}
+
+impl CloudflareKvBuilder {
+ /// Set the token used to authenticate with CloudFlare.
+ pub fn token(&mut self, token: &str) -> &mut Self {
+ if !token.is_empty() {
+ self.token = Some(token.to_string())
+ }
+ self
+ }
+
+ /// Set the account ID used to authenticate with CloudFlare.
+ pub fn account_id(&mut self, account_id: &str) -> &mut Self {
+ if !account_id.is_empty() {
+ self.account_id = Some(account_id.to_string())
+ }
+ self
+ }
+
+ /// Set the namespace ID.
+ pub fn namespace_id(&mut self, namespace_id: &str) -> &mut Self {
+ if !namespace_id.is_empty() {
+ self.namespace_id = Some(namespace_id.to_string())
+ }
+ self
+ }
+
+ /// Set the root within this backend.
+ pub fn root(&mut self, root: &str) -> &mut Self {
+ if !root.is_empty() {
+ self.root = Some(root.to_string())
+ }
+ self
+ }
+}
+
+impl Builder for CloudflareKvBuilder {
+ const SCHEME: Scheme = Scheme::CloudflareKv;
+
+ type Accessor = CloudflareKvBackend;
+
+ fn from_map(map: HashMap<String, String>) -> Self {
+ let mut builder = Self::default();
+ map.get("token").map(|v| builder.token(v));
+ map.get("account_id").map(|v| builder.account_id(v));
+ map.get("namespace_id").map(|v| builder.namespace_id(v));
+ map.get("root").map(|v| builder.root(v));
+ builder
+ }
+
+ fn build(&mut self) -> Result<Self::Accessor> {
+ let mut authorization = None;
+ if let Some(token) = &self.token {
+ authorization = Some(format_authorization_by_bearer(token)?)
+ }
+
+ let Some(account_id) = self.account_id.clone() else {
+ return Err(Error::new(
+ ErrorKind::ConfigInvalid,
+ "account_id is required",
+ ));
+ };
+
+ let Some(namespace_id) = self.namespace_id.clone() else {
+ return Err(Error::new(
+ ErrorKind::ConfigInvalid,
+ "namespace_id is required",
+ ));
+ };
+
+ 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::CloudflareKv)
+ })?
+ };
+
+ let root = normalize_root(
+ self.root
+ .clone()
+ .unwrap_or_else(|| "/".to_string())
+ .as_str(),
+ );
+
+ Ok(kv::Backend::new(Adapter {
+ authorization,
+ account_id,
+ namespace_id,
+ client,
+ })
+ .with_root(&root))
+ }
+}
+
+pub type CloudflareKvBackend = kv::Backend<Adapter>;
+
+#[derive(Clone)]
+pub struct Adapter {
+ authorization: Option<String>,
+ account_id: String,
+ namespace_id: String,
+ client: HttpClient,
+}
+
+impl Debug for Adapter {
+ fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+ f.debug_struct("Adapter")
+ .field("account_id", &self.account_id)
+ .field("namespace_id", &self.namespace_id)
+ .finish()
+ }
+}
+
+impl Adapter {
+ fn load_token<T>(&self, mut req: Request<T>) -> Result<Request<T>> {
+ if let Some(auth) = &self.authorization {
+ req.headers_mut()
+ .insert(header::AUTHORIZATION, auth.parse().unwrap());
+ }
+ Ok(req)
+ }
+
+ #[inline]
+ pub async fn send(&self, req: Request<AsyncBody>) -> Result<Response<IncomingAsyncBody>> {
Review Comment:
This wrapper seems not needed?
##########
core/src/services/cloudflare_kv/backend.rs:
##########
@@ -0,0 +1,444 @@
+// 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.
+#![allow(dead_code)]
+use std::collections::HashMap;
+use std::fmt::Debug;
+use std::fmt::Formatter;
+
+use async_trait::async_trait;
+use http::header;
+use http::Request;
+use http::Response;
+use http::StatusCode;
+use serde::Deserialize;
+
+use super::error::parse_error;
+use crate::raw::adapters::kv;
+use crate::raw::*;
+use crate::ErrorKind;
+use crate::*;
+
+#[doc = include_str!("docs.md")]
+#[derive(Default)]
+pub struct CloudflareKvBuilder {
+ /// The token used to authenticate with CloudFlare.
+ token: Option<String>,
+ /// The account ID used to authenticate with CloudFlare. Used as URI path parameter.
+ account_id: Option<String>,
+ /// The namespace ID. Used as URI path parameter.
+ namespace_id: Option<String>,
+
+ /// The HTTP client used to communicate with CloudFlare.
+ http_client: Option<HttpClient>,
+ /// Root within this backend.
+ root: Option<String>,
+}
+
+impl Debug for CloudflareKvBuilder {
+ fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+ f.debug_struct("CloudFlareKvBuilder")
+ .field("account_id", &self.account_id)
+ .field("namespace_id", &self.namespace_id)
+ .field("root", &self.root)
+ .finish()
+ }
+}
+
+impl CloudflareKvBuilder {
+ /// Set the token used to authenticate with CloudFlare.
+ pub fn token(&mut self, token: &str) -> &mut Self {
+ if !token.is_empty() {
+ self.token = Some(token.to_string())
+ }
+ self
+ }
+
+ /// Set the account ID used to authenticate with CloudFlare.
+ pub fn account_id(&mut self, account_id: &str) -> &mut Self {
+ if !account_id.is_empty() {
+ self.account_id = Some(account_id.to_string())
+ }
+ self
+ }
+
+ /// Set the namespace ID.
+ pub fn namespace_id(&mut self, namespace_id: &str) -> &mut Self {
+ if !namespace_id.is_empty() {
+ self.namespace_id = Some(namespace_id.to_string())
+ }
+ self
+ }
+
+ /// Set the root within this backend.
+ pub fn root(&mut self, root: &str) -> &mut Self {
+ if !root.is_empty() {
+ self.root = Some(root.to_string())
+ }
+ self
+ }
+}
+
+impl Builder for CloudflareKvBuilder {
+ const SCHEME: Scheme = Scheme::CloudflareKv;
+
+ type Accessor = CloudflareKvBackend;
+
+ fn from_map(map: HashMap<String, String>) -> Self {
+ let mut builder = Self::default();
+ map.get("token").map(|v| builder.token(v));
+ map.get("account_id").map(|v| builder.account_id(v));
+ map.get("namespace_id").map(|v| builder.namespace_id(v));
+ map.get("root").map(|v| builder.root(v));
+ builder
+ }
+
+ fn build(&mut self) -> Result<Self::Accessor> {
+ let mut authorization = None;
+ if let Some(token) = &self.token {
+ authorization = Some(format_authorization_by_bearer(token)?)
+ }
+
+ let Some(account_id) = self.account_id.clone() else {
+ return Err(Error::new(
+ ErrorKind::ConfigInvalid,
+ "account_id is required",
+ ));
+ };
+
+ let Some(namespace_id) = self.namespace_id.clone() else {
+ return Err(Error::new(
+ ErrorKind::ConfigInvalid,
+ "namespace_id is required",
+ ));
+ };
+
+ 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::CloudflareKv)
+ })?
+ };
+
+ let root = normalize_root(
+ self.root
+ .clone()
+ .unwrap_or_else(|| "/".to_string())
+ .as_str(),
+ );
+
+ Ok(kv::Backend::new(Adapter {
+ authorization,
+ account_id,
+ namespace_id,
+ client,
+ })
+ .with_root(&root))
+ }
+}
+
+pub type CloudflareKvBackend = kv::Backend<Adapter>;
+
+#[derive(Clone)]
+pub struct Adapter {
+ authorization: Option<String>,
+ account_id: String,
+ namespace_id: String,
+ client: HttpClient,
+}
+
+impl Debug for Adapter {
+ fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+ f.debug_struct("Adapter")
+ .field("account_id", &self.account_id)
+ .field("namespace_id", &self.namespace_id)
+ .finish()
+ }
+}
+
+impl Adapter {
+ fn load_token<T>(&self, mut req: Request<T>) -> Result<Request<T>> {
+ if let Some(auth) = &self.authorization {
+ req.headers_mut()
+ .insert(header::AUTHORIZATION, auth.parse().unwrap());
+ }
+ Ok(req)
+ }
+
+ #[inline]
+ pub async fn send(&self, req: Request<AsyncBody>) -> Result<Response<IncomingAsyncBody>> {
+ self.client.send(req).await
+ }
+}
+
+#[async_trait]
+impl kv::Adapter for Adapter {
+ fn metadata(&self) -> kv::Metadata {
+ kv::Metadata::new(
+ Scheme::CloudflareKv,
+ &self.namespace_id,
+ Capability {
+ read: true,
+ write: true,
+ list: true,
+
+ ..Default::default()
+ },
+ )
+ }
+
+ async fn get(&self, path: &str) -> Result<Option<Vec<u8>>> {
+ let url = format!(
+ r"https://api.cloudflare.com/client/v4/accounts/{}/storage/kv/namespaces/{}",
+ self.account_id, self.namespace_id
+ );
+ let url = format!("{}/values/{}", url, path);
+ let mut req = Request::get(&url);
+ req = req.header(header::CONTENT_TYPE, "application/json");
+ let mut req = req
+ .body(AsyncBody::Empty)
+ .map_err(new_request_build_error)?;
+ req = self.load_token(req)?;
+ let resp = self.client.send(req).await?;
+ let status = resp.status();
+ match status {
+ StatusCode::OK => {
+ let body = resp.into_body().bytes().await?;
+ Ok(Some(body.into()))
+ }
+ _ => Err(parse_error(resp).await?),
+ }
+ }
+
+ async fn set(&self, path: &str, value: &[u8]) -> Result<()> {
+ let url = format!(
+ r"https://api.cloudflare.com/client/v4/accounts/{}/storage/kv/namespaces/{}",
+ self.account_id, self.namespace_id
+ );
+ let url = format!("{}/values/{}", url, path);
+ let mut req = Request::put(&url);
+ req = req.header(header::CONTENT_TYPE, "multipart/form-data");
+ let multipart = Multipart::new();
+ let multipart = multipart
+ .part(FormDataPart::new("metadata").content(serde_json::Value::Null.to_string()))
+ .part(FormDataPart::new("value").content(value.to_vec()));
+ let mut req = multipart.apply(req)?;
+ req = self.load_token(req)?;
+ let resp = self.client.send(req).await?;
+ let status = resp.status();
+ match status {
+ StatusCode::OK => {
+ let body = resp.into_body().bytes().await?;
+ let response: CfKvResponse = serde_json::from_slice(&body).map_err(|e| {
Review Comment:
cloudflare makes sure that `success` is true while status code is Ok, maybe we can ignore those logic?
##########
core/src/services/cloudflare_kv/backend.rs:
##########
@@ -0,0 +1,444 @@
+// 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.
+#![allow(dead_code)]
+use std::collections::HashMap;
+use std::fmt::Debug;
+use std::fmt::Formatter;
+
+use async_trait::async_trait;
+use http::header;
+use http::Request;
+use http::Response;
+use http::StatusCode;
+use serde::Deserialize;
+
+use super::error::parse_error;
+use crate::raw::adapters::kv;
+use crate::raw::*;
+use crate::ErrorKind;
+use crate::*;
+
+#[doc = include_str!("docs.md")]
+#[derive(Default)]
+pub struct CloudflareKvBuilder {
+ /// The token used to authenticate with CloudFlare.
+ token: Option<String>,
+ /// The account ID used to authenticate with CloudFlare. Used as URI path parameter.
+ account_id: Option<String>,
+ /// The namespace ID. Used as URI path parameter.
+ namespace_id: Option<String>,
+
+ /// The HTTP client used to communicate with CloudFlare.
+ http_client: Option<HttpClient>,
+ /// Root within this backend.
+ root: Option<String>,
+}
+
+impl Debug for CloudflareKvBuilder {
+ fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+ f.debug_struct("CloudFlareKvBuilder")
+ .field("account_id", &self.account_id)
+ .field("namespace_id", &self.namespace_id)
+ .field("root", &self.root)
+ .finish()
+ }
+}
+
+impl CloudflareKvBuilder {
+ /// Set the token used to authenticate with CloudFlare.
+ pub fn token(&mut self, token: &str) -> &mut Self {
+ if !token.is_empty() {
+ self.token = Some(token.to_string())
+ }
+ self
+ }
+
+ /// Set the account ID used to authenticate with CloudFlare.
+ pub fn account_id(&mut self, account_id: &str) -> &mut Self {
+ if !account_id.is_empty() {
+ self.account_id = Some(account_id.to_string())
+ }
+ self
+ }
+
+ /// Set the namespace ID.
+ pub fn namespace_id(&mut self, namespace_id: &str) -> &mut Self {
+ if !namespace_id.is_empty() {
+ self.namespace_id = Some(namespace_id.to_string())
+ }
+ self
+ }
+
+ /// Set the root within this backend.
+ pub fn root(&mut self, root: &str) -> &mut Self {
+ if !root.is_empty() {
+ self.root = Some(root.to_string())
+ }
+ self
+ }
+}
+
+impl Builder for CloudflareKvBuilder {
+ const SCHEME: Scheme = Scheme::CloudflareKv;
+
+ type Accessor = CloudflareKvBackend;
+
+ fn from_map(map: HashMap<String, String>) -> Self {
+ let mut builder = Self::default();
+ map.get("token").map(|v| builder.token(v));
+ map.get("account_id").map(|v| builder.account_id(v));
+ map.get("namespace_id").map(|v| builder.namespace_id(v));
+ map.get("root").map(|v| builder.root(v));
+ builder
+ }
+
+ fn build(&mut self) -> Result<Self::Accessor> {
+ let mut authorization = None;
+ if let Some(token) = &self.token {
+ authorization = Some(format_authorization_by_bearer(token)?)
+ }
+
+ let Some(account_id) = self.account_id.clone() else {
+ return Err(Error::new(
+ ErrorKind::ConfigInvalid,
+ "account_id is required",
+ ));
+ };
+
+ let Some(namespace_id) = self.namespace_id.clone() else {
+ return Err(Error::new(
+ ErrorKind::ConfigInvalid,
+ "namespace_id is required",
+ ));
+ };
+
+ 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::CloudflareKv)
+ })?
+ };
+
+ let root = normalize_root(
+ self.root
+ .clone()
+ .unwrap_or_else(|| "/".to_string())
+ .as_str(),
+ );
+
+ Ok(kv::Backend::new(Adapter {
+ authorization,
+ account_id,
+ namespace_id,
+ client,
+ })
+ .with_root(&root))
+ }
+}
+
+pub type CloudflareKvBackend = kv::Backend<Adapter>;
+
+#[derive(Clone)]
+pub struct Adapter {
+ authorization: Option<String>,
+ account_id: String,
+ namespace_id: String,
+ client: HttpClient,
+}
+
+impl Debug for Adapter {
+ fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+ f.debug_struct("Adapter")
+ .field("account_id", &self.account_id)
+ .field("namespace_id", &self.namespace_id)
+ .finish()
+ }
+}
+
+impl Adapter {
+ fn load_token<T>(&self, mut req: Request<T>) -> Result<Request<T>> {
+ if let Some(auth) = &self.authorization {
+ req.headers_mut()
+ .insert(header::AUTHORIZATION, auth.parse().unwrap());
+ }
+ Ok(req)
+ }
+
+ #[inline]
+ pub async fn send(&self, req: Request<AsyncBody>) -> Result<Response<IncomingAsyncBody>> {
+ self.client.send(req).await
+ }
+}
+
+#[async_trait]
+impl kv::Adapter for Adapter {
+ fn metadata(&self) -> kv::Metadata {
+ kv::Metadata::new(
+ Scheme::CloudflareKv,
+ &self.namespace_id,
+ Capability {
+ read: true,
+ write: true,
+ list: true,
+
+ ..Default::default()
+ },
+ )
+ }
+
+ async fn get(&self, path: &str) -> Result<Option<Vec<u8>>> {
+ let url = format!(
+ r"https://api.cloudflare.com/client/v4/accounts/{}/storage/kv/namespaces/{}",
+ self.account_id, self.namespace_id
+ );
+ let url = format!("{}/values/{}", url, path);
+ let mut req = Request::get(&url);
+ req = req.header(header::CONTENT_TYPE, "application/json");
+ let mut req = req
+ .body(AsyncBody::Empty)
+ .map_err(new_request_build_error)?;
+ req = self.load_token(req)?;
+ let resp = self.client.send(req).await?;
+ let status = resp.status();
+ match status {
+ StatusCode::OK => {
+ let body = resp.into_body().bytes().await?;
+ Ok(Some(body.into()))
+ }
+ _ => Err(parse_error(resp).await?),
+ }
+ }
+
+ async fn set(&self, path: &str, value: &[u8]) -> Result<()> {
+ let url = format!(
+ r"https://api.cloudflare.com/client/v4/accounts/{}/storage/kv/namespaces/{}",
+ self.account_id, self.namespace_id
+ );
+ let url = format!("{}/values/{}", url, path);
+ let mut req = Request::put(&url);
+ req = req.header(header::CONTENT_TYPE, "multipart/form-data");
+ let multipart = Multipart::new();
+ let multipart = multipart
+ .part(FormDataPart::new("metadata").content(serde_json::Value::Null.to_string()))
+ .part(FormDataPart::new("value").content(value.to_vec()));
+ let mut req = multipart.apply(req)?;
+ req = self.load_token(req)?;
+ let resp = self.client.send(req).await?;
+ let status = resp.status();
+ match status {
+ StatusCode::OK => {
+ let body = resp.into_body().bytes().await?;
+ let response: CfKvResponse = serde_json::from_slice(&body).map_err(|e| {
+ Error::new(
+ crate::ErrorKind::Unexpected,
+ &format!("failed to parse error response: {}", e),
+ )
+ })?;
+ if !response.success {
+ return Err(Error::new(
+ crate::ErrorKind::Unexpected,
+ &String::from_utf8_lossy(&body),
+ ));
+ }
+ Ok(())
+ }
+ _ => Err(parse_error(resp).await?),
+ }
+ }
+
+ async fn delete(&self, path: &str) -> Result<()> {
+ let url = format!(
+ r"https://api.cloudflare.com/client/v4/accounts/{}/storage/kv/namespaces/{}",
+ self.account_id, self.namespace_id
+ );
+ let url = format!("{}/values/{}", url, path);
+ let mut req = Request::delete(&url);
+ req = req.header(header::CONTENT_TYPE, "application/json");
+ let mut req = req
+ .body(AsyncBody::Empty)
+ .map_err(new_request_build_error)?;
+ req = self.load_token(req)?;
+ let resp = self.client.send(req).await?;
+ let status = resp.status();
+ match status {
+ StatusCode::OK => {
+ let body = resp.into_body().bytes().await?;
Review Comment:
The same
##########
core/src/services/cloudflare_kv/backend.rs:
##########
@@ -0,0 +1,444 @@
+// 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.
+#![allow(dead_code)]
+use std::collections::HashMap;
+use std::fmt::Debug;
+use std::fmt::Formatter;
+
+use async_trait::async_trait;
+use http::header;
+use http::Request;
+use http::Response;
+use http::StatusCode;
+use serde::Deserialize;
+
+use super::error::parse_error;
+use crate::raw::adapters::kv;
+use crate::raw::*;
+use crate::ErrorKind;
+use crate::*;
+
+#[doc = include_str!("docs.md")]
+#[derive(Default)]
+pub struct CloudflareKvBuilder {
+ /// The token used to authenticate with CloudFlare.
+ token: Option<String>,
+ /// The account ID used to authenticate with CloudFlare. Used as URI path parameter.
+ account_id: Option<String>,
+ /// The namespace ID. Used as URI path parameter.
+ namespace_id: Option<String>,
+
+ /// The HTTP client used to communicate with CloudFlare.
+ http_client: Option<HttpClient>,
+ /// Root within this backend.
+ root: Option<String>,
+}
+
+impl Debug for CloudflareKvBuilder {
+ fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+ f.debug_struct("CloudFlareKvBuilder")
+ .field("account_id", &self.account_id)
+ .field("namespace_id", &self.namespace_id)
+ .field("root", &self.root)
+ .finish()
+ }
+}
+
+impl CloudflareKvBuilder {
+ /// Set the token used to authenticate with CloudFlare.
+ pub fn token(&mut self, token: &str) -> &mut Self {
+ if !token.is_empty() {
+ self.token = Some(token.to_string())
+ }
+ self
+ }
+
+ /// Set the account ID used to authenticate with CloudFlare.
+ pub fn account_id(&mut self, account_id: &str) -> &mut Self {
+ if !account_id.is_empty() {
+ self.account_id = Some(account_id.to_string())
+ }
+ self
+ }
+
+ /// Set the namespace ID.
+ pub fn namespace_id(&mut self, namespace_id: &str) -> &mut Self {
+ if !namespace_id.is_empty() {
+ self.namespace_id = Some(namespace_id.to_string())
+ }
+ self
+ }
+
+ /// Set the root within this backend.
+ pub fn root(&mut self, root: &str) -> &mut Self {
+ if !root.is_empty() {
+ self.root = Some(root.to_string())
+ }
+ self
+ }
+}
+
+impl Builder for CloudflareKvBuilder {
+ const SCHEME: Scheme = Scheme::CloudflareKv;
+
+ type Accessor = CloudflareKvBackend;
+
+ fn from_map(map: HashMap<String, String>) -> Self {
+ let mut builder = Self::default();
+ map.get("token").map(|v| builder.token(v));
+ map.get("account_id").map(|v| builder.account_id(v));
+ map.get("namespace_id").map(|v| builder.namespace_id(v));
+ map.get("root").map(|v| builder.root(v));
+ builder
+ }
+
+ fn build(&mut self) -> Result<Self::Accessor> {
+ let mut authorization = None;
+ if let Some(token) = &self.token {
+ authorization = Some(format_authorization_by_bearer(token)?)
+ }
+
+ let Some(account_id) = self.account_id.clone() else {
+ return Err(Error::new(
+ ErrorKind::ConfigInvalid,
+ "account_id is required",
+ ));
+ };
+
+ let Some(namespace_id) = self.namespace_id.clone() else {
+ return Err(Error::new(
+ ErrorKind::ConfigInvalid,
+ "namespace_id is required",
+ ));
+ };
+
+ 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::CloudflareKv)
+ })?
+ };
+
+ let root = normalize_root(
+ self.root
+ .clone()
+ .unwrap_or_else(|| "/".to_string())
+ .as_str(),
+ );
+
+ Ok(kv::Backend::new(Adapter {
+ authorization,
+ account_id,
+ namespace_id,
+ client,
+ })
+ .with_root(&root))
+ }
+}
+
+pub type CloudflareKvBackend = kv::Backend<Adapter>;
+
+#[derive(Clone)]
+pub struct Adapter {
+ authorization: Option<String>,
+ account_id: String,
+ namespace_id: String,
+ client: HttpClient,
+}
+
+impl Debug for Adapter {
+ fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+ f.debug_struct("Adapter")
+ .field("account_id", &self.account_id)
+ .field("namespace_id", &self.namespace_id)
+ .finish()
+ }
+}
+
+impl Adapter {
+ fn load_token<T>(&self, mut req: Request<T>) -> Result<Request<T>> {
+ if let Some(auth) = &self.authorization {
+ req.headers_mut()
+ .insert(header::AUTHORIZATION, auth.parse().unwrap());
+ }
+ Ok(req)
+ }
+
+ #[inline]
+ pub async fn send(&self, req: Request<AsyncBody>) -> Result<Response<IncomingAsyncBody>> {
+ self.client.send(req).await
+ }
+}
+
+#[async_trait]
+impl kv::Adapter for Adapter {
+ fn metadata(&self) -> kv::Metadata {
+ kv::Metadata::new(
+ Scheme::CloudflareKv,
+ &self.namespace_id,
+ Capability {
+ read: true,
+ write: true,
+ list: true,
+
+ ..Default::default()
+ },
+ )
+ }
+
+ async fn get(&self, path: &str) -> Result<Option<Vec<u8>>> {
+ let url = format!(
+ r"https://api.cloudflare.com/client/v4/accounts/{}/storage/kv/namespaces/{}",
+ self.account_id, self.namespace_id
+ );
+ let url = format!("{}/values/{}", url, path);
+ let mut req = Request::get(&url);
+ req = req.header(header::CONTENT_TYPE, "application/json");
+ let mut req = req
+ .body(AsyncBody::Empty)
+ .map_err(new_request_build_error)?;
+ req = self.load_token(req)?;
+ let resp = self.client.send(req).await?;
+ let status = resp.status();
+ match status {
+ StatusCode::OK => {
+ let body = resp.into_body().bytes().await?;
+ Ok(Some(body.into()))
+ }
+ _ => Err(parse_error(resp).await?),
+ }
+ }
+
+ async fn set(&self, path: &str, value: &[u8]) -> Result<()> {
+ let url = format!(
+ r"https://api.cloudflare.com/client/v4/accounts/{}/storage/kv/namespaces/{}",
Review Comment:
Maybe we can extract `https://api.cloudflare.com/client/v4` as a constant? Or can store the url `https://api.cloudflare.com/client/v4/accounts/{account_id}/storage/kv/namespaces/{namespace_id}` in adapter instead of build it everytime.
##########
core/src/services/cloudflare_kv/backend.rs:
##########
@@ -0,0 +1,444 @@
+// 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.
+#![allow(dead_code)]
+use std::collections::HashMap;
+use std::fmt::Debug;
+use std::fmt::Formatter;
+
+use async_trait::async_trait;
+use http::header;
+use http::Request;
+use http::Response;
+use http::StatusCode;
+use serde::Deserialize;
+
+use super::error::parse_error;
+use crate::raw::adapters::kv;
+use crate::raw::*;
+use crate::ErrorKind;
+use crate::*;
+
+#[doc = include_str!("docs.md")]
+#[derive(Default)]
+pub struct CloudflareKvBuilder {
+ /// The token used to authenticate with CloudFlare.
+ token: Option<String>,
+ /// The account ID used to authenticate with CloudFlare. Used as URI path parameter.
+ account_id: Option<String>,
+ /// The namespace ID. Used as URI path parameter.
+ namespace_id: Option<String>,
+
+ /// The HTTP client used to communicate with CloudFlare.
+ http_client: Option<HttpClient>,
+ /// Root within this backend.
+ root: Option<String>,
+}
+
+impl Debug for CloudflareKvBuilder {
+ fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+ f.debug_struct("CloudFlareKvBuilder")
+ .field("account_id", &self.account_id)
+ .field("namespace_id", &self.namespace_id)
+ .field("root", &self.root)
+ .finish()
+ }
+}
+
+impl CloudflareKvBuilder {
+ /// Set the token used to authenticate with CloudFlare.
+ pub fn token(&mut self, token: &str) -> &mut Self {
+ if !token.is_empty() {
+ self.token = Some(token.to_string())
+ }
+ self
+ }
+
+ /// Set the account ID used to authenticate with CloudFlare.
+ pub fn account_id(&mut self, account_id: &str) -> &mut Self {
+ if !account_id.is_empty() {
+ self.account_id = Some(account_id.to_string())
+ }
+ self
+ }
+
+ /// Set the namespace ID.
+ pub fn namespace_id(&mut self, namespace_id: &str) -> &mut Self {
+ if !namespace_id.is_empty() {
+ self.namespace_id = Some(namespace_id.to_string())
+ }
+ self
+ }
+
+ /// Set the root within this backend.
+ pub fn root(&mut self, root: &str) -> &mut Self {
+ if !root.is_empty() {
+ self.root = Some(root.to_string())
+ }
+ self
+ }
+}
+
+impl Builder for CloudflareKvBuilder {
+ const SCHEME: Scheme = Scheme::CloudflareKv;
+
+ type Accessor = CloudflareKvBackend;
+
+ fn from_map(map: HashMap<String, String>) -> Self {
+ let mut builder = Self::default();
+ map.get("token").map(|v| builder.token(v));
+ map.get("account_id").map(|v| builder.account_id(v));
+ map.get("namespace_id").map(|v| builder.namespace_id(v));
+ map.get("root").map(|v| builder.root(v));
+ builder
+ }
+
+ fn build(&mut self) -> Result<Self::Accessor> {
+ let mut authorization = None;
+ if let Some(token) = &self.token {
+ authorization = Some(format_authorization_by_bearer(token)?)
+ }
+
+ let Some(account_id) = self.account_id.clone() else {
+ return Err(Error::new(
+ ErrorKind::ConfigInvalid,
+ "account_id is required",
+ ));
+ };
+
+ let Some(namespace_id) = self.namespace_id.clone() else {
+ return Err(Error::new(
+ ErrorKind::ConfigInvalid,
+ "namespace_id is required",
+ ));
+ };
+
+ 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::CloudflareKv)
+ })?
+ };
+
+ let root = normalize_root(
+ self.root
+ .clone()
+ .unwrap_or_else(|| "/".to_string())
+ .as_str(),
+ );
+
+ Ok(kv::Backend::new(Adapter {
+ authorization,
+ account_id,
+ namespace_id,
+ client,
+ })
+ .with_root(&root))
+ }
+}
+
+pub type CloudflareKvBackend = kv::Backend<Adapter>;
+
+#[derive(Clone)]
+pub struct Adapter {
+ authorization: Option<String>,
+ account_id: String,
+ namespace_id: String,
+ client: HttpClient,
+}
+
+impl Debug for Adapter {
+ fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+ f.debug_struct("Adapter")
+ .field("account_id", &self.account_id)
+ .field("namespace_id", &self.namespace_id)
+ .finish()
+ }
+}
+
+impl Adapter {
+ fn load_token<T>(&self, mut req: Request<T>) -> Result<Request<T>> {
+ if let Some(auth) = &self.authorization {
+ req.headers_mut()
+ .insert(header::AUTHORIZATION, auth.parse().unwrap());
+ }
+ Ok(req)
+ }
+
+ #[inline]
+ pub async fn send(&self, req: Request<AsyncBody>) -> Result<Response<IncomingAsyncBody>> {
+ self.client.send(req).await
+ }
+}
+
+#[async_trait]
+impl kv::Adapter for Adapter {
+ fn metadata(&self) -> kv::Metadata {
+ kv::Metadata::new(
+ Scheme::CloudflareKv,
+ &self.namespace_id,
+ Capability {
+ read: true,
+ write: true,
+ list: true,
+
+ ..Default::default()
+ },
+ )
+ }
+
+ async fn get(&self, path: &str) -> Result<Option<Vec<u8>>> {
+ let url = format!(
+ r"https://api.cloudflare.com/client/v4/accounts/{}/storage/kv/namespaces/{}",
+ self.account_id, self.namespace_id
+ );
+ let url = format!("{}/values/{}", url, path);
+ let mut req = Request::get(&url);
+ req = req.header(header::CONTENT_TYPE, "application/json");
+ let mut req = req
+ .body(AsyncBody::Empty)
+ .map_err(new_request_build_error)?;
+ req = self.load_token(req)?;
+ let resp = self.client.send(req).await?;
+ let status = resp.status();
+ match status {
+ StatusCode::OK => {
+ let body = resp.into_body().bytes().await?;
+ Ok(Some(body.into()))
+ }
+ _ => Err(parse_error(resp).await?),
+ }
+ }
+
+ async fn set(&self, path: &str, value: &[u8]) -> Result<()> {
+ let url = format!(
+ r"https://api.cloudflare.com/client/v4/accounts/{}/storage/kv/namespaces/{}",
+ self.account_id, self.namespace_id
+ );
+ let url = format!("{}/values/{}", url, path);
+ let mut req = Request::put(&url);
+ req = req.header(header::CONTENT_TYPE, "multipart/form-data");
+ let multipart = Multipart::new();
+ let multipart = multipart
+ .part(FormDataPart::new("metadata").content(serde_json::Value::Null.to_string()))
+ .part(FormDataPart::new("value").content(value.to_vec()));
+ let mut req = multipart.apply(req)?;
+ req = self.load_token(req)?;
+ let resp = self.client.send(req).await?;
+ let status = resp.status();
+ match status {
+ StatusCode::OK => {
+ let body = resp.into_body().bytes().await?;
+ let response: CfKvResponse = serde_json::from_slice(&body).map_err(|e| {
+ Error::new(
+ crate::ErrorKind::Unexpected,
+ &format!("failed to parse error response: {}", e),
+ )
+ })?;
+ if !response.success {
+ return Err(Error::new(
+ crate::ErrorKind::Unexpected,
+ &String::from_utf8_lossy(&body),
+ ));
+ }
+ Ok(())
+ }
+ _ => Err(parse_error(resp).await?),
+ }
+ }
+
+ async fn delete(&self, path: &str) -> Result<()> {
+ let url = format!(
+ r"https://api.cloudflare.com/client/v4/accounts/{}/storage/kv/namespaces/{}",
+ self.account_id, self.namespace_id
+ );
+ let url = format!("{}/values/{}", url, path);
+ let mut req = Request::delete(&url);
+ req = req.header(header::CONTENT_TYPE, "application/json");
+ let mut req = req
+ .body(AsyncBody::Empty)
+ .map_err(new_request_build_error)?;
+ req = self.load_token(req)?;
+ let resp = self.client.send(req).await?;
+ let status = resp.status();
+ match status {
+ StatusCode::OK => {
+ let body = resp.into_body().bytes().await?;
+ let response: CfKvResponse = serde_json::from_slice(&body).map_err(|e| {
+ Error::new(
+ crate::ErrorKind::Unexpected,
+ &format!("failed to parse error response: {}", e),
+ )
+ })?;
+ if !response.success {
+ return Err(Error::new(
+ crate::ErrorKind::Unexpected,
+ &String::from_utf8_lossy(&body),
+ ));
+ }
+ Ok(())
+ }
+ _ => Err(parse_error(resp).await?),
+ }
+ }
+
+ async fn scan(&self, path: &str) -> Result<Vec<String>> {
+ let url = format!(
+ r"https://api.cloudflare.com/client/v4/accounts/{}/storage/kv/namespaces/{}",
+ self.account_id, self.namespace_id
+ );
+ let mut url = format!("{}/keys", url);
+ if path.len() > 0 {
+ url = format!("{}?prefix={}", url, path);
+ }
+ let mut req = Request::get(&url);
+ req = req.header(header::CONTENT_TYPE, "application/json");
+ let mut req = req
+ .body(AsyncBody::Empty)
+ .map_err(new_request_build_error)?;
+ req = self.load_token(req)?;
+ let resp = self.client.send(req).await?;
+ let status = resp.status();
+ match status {
+ StatusCode::OK => {
+ let body = resp.into_body().bytes().await?;
+ let response: CfKvScanResponse = serde_json::from_slice(&body).map_err(|e| {
Review Comment:
We should continuously send cursor requests until we retrieve the entire content. Improvements can be made in the next PR or through a new kv API.
--
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