opendal_service_d1/
backend.rs

1// Licensed to the Apache Software Foundation (ASF) under one
2// or more contributor license agreements.  See the NOTICE file
3// distributed with this work for additional information
4// regarding copyright ownership.  The ASF licenses this file
5// to you under the Apache License, Version 2.0 (the
6// "License"); you may not use this file except in compliance
7// with the License.  You may obtain a copy of the License at
8//
9//   http://www.apache.org/licenses/LICENSE-2.0
10//
11// Unless required by applicable law or agreed to in writing,
12// software distributed under the License is distributed on an
13// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
14// KIND, either express or implied.  See the License for the
15// specific language governing permissions and limitations
16// under the License.
17
18use std::fmt::Debug;
19use std::sync::Arc;
20
21use super::D1_SCHEME;
22use super::config::D1Config;
23use super::core::*;
24use super::deleter::D1Deleter;
25use super::reader::*;
26use super::writer::D1Writer;
27use opendal_core::raw::*;
28use opendal_core::*;
29
30#[doc = include_str!("docs.md")]
31#[derive(Default)]
32pub struct D1Builder {
33    pub(super) config: D1Config,
34}
35
36impl Debug for D1Builder {
37    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
38        f.debug_struct("D1Builder")
39            .field("config", &self.config)
40            .finish_non_exhaustive()
41    }
42}
43
44impl D1Builder {
45    /// Set api token for the cloudflare d1 service.
46    ///
47    /// create a api token from [here](https://dash.cloudflare.com/profile/api-tokens)
48    pub fn token(mut self, token: &str) -> Self {
49        if !token.is_empty() {
50            self.config.token = Some(token.to_string());
51        }
52        self
53    }
54
55    /// Set the account identifier for the cloudflare d1 service.
56    ///
57    /// get the account identifier from Workers & Pages -> Overview -> Account ID
58    /// If not specified, it will return an error when building.
59    pub fn account_id(mut self, account_id: &str) -> Self {
60        if !account_id.is_empty() {
61            self.config.account_id = Some(account_id.to_string());
62        }
63        self
64    }
65
66    /// Set the database identifier for the cloudflare d1 service.
67    ///
68    /// get the database identifier from Workers & Pages -> D1 -> [Your Database] -> Database ID
69    /// If not specified, it will return an error when building.
70    pub fn database_id(mut self, database_id: &str) -> Self {
71        if !database_id.is_empty() {
72            self.config.database_id = Some(database_id.to_string());
73        }
74        self
75    }
76
77    /// set the working directory, all operations will be performed under it.
78    ///
79    /// default: "/"
80    pub fn root(mut self, root: &str) -> Self {
81        self.config.root = if root.is_empty() {
82            None
83        } else {
84            Some(root.to_string())
85        };
86
87        self
88    }
89
90    /// Set the table name of the d1 service to read/write.
91    ///
92    /// If not specified, it will return an error when building.
93    pub fn table(mut self, table: &str) -> Self {
94        if !table.is_empty() {
95            self.config.table = Some(table.to_owned());
96        }
97        self
98    }
99
100    /// Set the key field name of the d1 service to read/write.
101    ///
102    /// Default to `key` if not specified.
103    pub fn key_field(mut self, key_field: &str) -> Self {
104        if !key_field.is_empty() {
105            self.config.key_field = Some(key_field.to_string());
106        }
107        self
108    }
109
110    /// Set the value field name of the d1 service to read/write.
111    ///
112    /// Default to `value` if not specified.
113    pub fn value_field(mut self, value_field: &str) -> Self {
114        if !value_field.is_empty() {
115            self.config.value_field = Some(value_field.to_string());
116        }
117        self
118    }
119}
120
121impl Builder for D1Builder {
122    type Config = D1Config;
123
124    fn build(self) -> Result<impl Service> {
125        let mut authorization = None;
126        let config = self.config;
127
128        if let Some(token) = config.token {
129            authorization = Some(format_authorization_by_bearer(&token)?)
130        }
131
132        let Some(account_id) = config.account_id else {
133            return Err(Error::new(
134                ErrorKind::ConfigInvalid,
135                "account_id is required",
136            ));
137        };
138
139        let Some(database_id) = config.database_id.clone() else {
140            return Err(Error::new(
141                ErrorKind::ConfigInvalid,
142                "database_id is required",
143            ));
144        };
145
146        let Some(table) = config.table.clone() else {
147            return Err(Error::new(ErrorKind::ConfigInvalid, "table is required"));
148        };
149
150        let key_field = config
151            .key_field
152            .clone()
153            .unwrap_or_else(|| "key".to_string());
154
155        let value_field = config
156            .value_field
157            .clone()
158            .unwrap_or_else(|| "value".to_string());
159
160        let root = normalize_root(
161            config
162                .root
163                .clone()
164                .unwrap_or_else(|| "/".to_string())
165                .as_str(),
166        );
167        Ok(D1Backend::new(D1Core {
168            authorization,
169            account_id,
170            database_id,
171            table,
172            key_field,
173            value_field,
174        })
175        .with_normalized_root(root))
176    }
177}
178
179/// Backend for D1 services.
180#[derive(Clone, Debug)]
181pub struct D1Backend {
182    pub(crate) core: Arc<D1Core>,
183    pub(crate) root: String,
184    pub(crate) info: ServiceInfo,
185    pub(crate) capability: Capability,
186}
187
188impl D1Backend {
189    pub fn new(core: D1Core) -> Self {
190        let info = ServiceInfo::new(D1_SCHEME, "/", &core.table);
191        let capability = Capability {
192            read: true,
193            stat: true,
194            write: true,
195            write_can_empty: true,
196            // Cloudflare D1 supports 1MB as max in write_total.
197            // refer to https://developers.cloudflare.com/d1/platform/limits/
198            write_total_max_size: Some(1000 * 1000),
199            delete: true,
200            shared: true,
201            ..Default::default()
202        };
203
204        Self {
205            core: Arc::new(core),
206            root: "/".to_string(),
207            info,
208            capability,
209        }
210    }
211
212    fn with_normalized_root(mut self, root: String) -> Self {
213        self.info = self.info.with_root(&root);
214        self.root = root;
215        self
216    }
217}
218
219impl Service for D1Backend {
220    type Reader = oio::StreamReader<D1Reader>;
221    type Writer = D1Writer;
222    type Lister = ();
223    type Deleter = oio::OneShotDeleter<D1Deleter>;
224    type Copier = ();
225
226    fn info(&self) -> ServiceInfo {
227        self.info.clone()
228    }
229
230    fn capability(&self) -> Capability {
231        self.capability
232    }
233
234    async fn create_dir(
235        &self,
236        _ctx: &OperationContext,
237        _path: &str,
238        _args: OpCreateDir,
239    ) -> Result<RpCreateDir> {
240        Err(Error::new(
241            ErrorKind::Unsupported,
242            "operation is not supported",
243        ))
244    }
245
246    async fn stat(&self, ctx: &OperationContext, path: &str, _: OpStat) -> Result<RpStat> {
247        let p = build_abs_path(&self.root, path);
248
249        if p == build_abs_path(&self.root, "") {
250            Ok(RpStat::new(Metadata::new(EntryMode::DIR)))
251        } else {
252            match self.core.get_length(ctx, &p).await? {
253                Some(length) => Ok(RpStat::new(
254                    Metadata::new(EntryMode::FILE).with_content_length(length as u64),
255                )),
256                None => Err(Error::new(ErrorKind::NotFound, "kv not found in d1")),
257            }
258        }
259    }
260    fn read(&self, ctx: &OperationContext, path: &str, args: OpRead) -> Result<Self::Reader> {
261        let output: oio::StreamReader<D1Reader> = {
262            Ok(oio::StreamReader::new(D1Reader::new(
263                self.clone(),
264                ctx.clone(),
265                path,
266                args,
267            )))
268        }?;
269
270        Ok(output)
271    }
272
273    fn write(&self, ctx: &OperationContext, path: &str, _: OpWrite) -> Result<Self::Writer> {
274        let output: D1Writer = {
275            let p = build_abs_path(&self.root, path);
276            Ok(D1Writer::new(self.core.clone(), ctx.clone(), p))
277        }?;
278
279        Ok(output)
280    }
281
282    fn delete(&self, ctx: &OperationContext) -> Result<Self::Deleter> {
283        let output: oio::OneShotDeleter<D1Deleter> = {
284            Ok(oio::OneShotDeleter::new(D1Deleter::new(
285                self.core.clone(),
286                ctx.clone(),
287                self.root.clone(),
288            )))
289        }?;
290
291        Ok(output)
292    }
293
294    fn list(&self, _ctx: &OperationContext, _path: &str, _args: OpList) -> Result<Self::Lister> {
295        Err(Error::new(
296            ErrorKind::Unsupported,
297            "operation is not supported",
298        ))
299    }
300
301    fn copy(
302        &self,
303        _ctx: &OperationContext,
304        _from: &str,
305        _to: &str,
306        _args: OpCopy,
307        _opts: OpCopier,
308    ) -> Result<Self::Copier> {
309        Err(Error::new(
310            ErrorKind::Unsupported,
311            "operation is not supported",
312        ))
313    }
314
315    async fn rename(
316        &self,
317        _ctx: &OperationContext,
318        _from: &str,
319        _to: &str,
320        _args: OpRename,
321    ) -> Result<RpRename> {
322        Err(Error::new(
323            ErrorKind::Unsupported,
324            "operation is not supported",
325        ))
326    }
327
328    async fn presign(
329        &self,
330        _ctx: &OperationContext,
331        _path: &str,
332        _args: OpPresign,
333    ) -> Result<RpPresign> {
334        Err(Error::new(
335            ErrorKind::Unsupported,
336            "operation is not supported",
337        ))
338    }
339}
opendal_service_d1/backend.rs

opendal_service_d1/
backend.rs
