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    type Composer = ();
226
227    fn info(&self) -> ServiceInfo {
228        self.info.clone()
229    }
230
231    fn capability(&self) -> Capability {
232        self.capability
233    }
234
235    async fn create_dir(
236        &self,
237        _ctx: &OperationContext,
238        _path: &str,
239        _args: OpCreateDir,
240    ) -> Result<RpCreateDir> {
241        Err(Error::new(
242            ErrorKind::Unsupported,
243            "operation is not supported",
244        ))
245    }
246
247    async fn stat(&self, ctx: &OperationContext, path: &str, _: OpStat) -> Result<RpStat> {
248        let p = build_abs_path(&self.root, path);
249
250        if p == build_abs_path(&self.root, "") {
251            Ok(RpStat::new(MetadataBuilder::dir().build()))
252        } else {
253            match self.core.get_length(ctx, &p).await? {
254                Some(length) => Ok(RpStat::new({
255                    let metadata = MetadataBuilder::file(length as u64);
256                    metadata.build()
257                })),
258                None => Err(Error::new(ErrorKind::NotFound, "kv not found in d1")),
259            }
260        }
261    }
262    fn read(&self, ctx: &OperationContext, path: &str, args: OpRead) -> Result<Self::Reader> {
263        let output: oio::StreamReader<D1Reader> = {
264            Ok(oio::StreamReader::new(D1Reader::new(
265                self.clone(),
266                ctx.clone(),
267                path,
268                args,
269            )))
270        }?;
271
272        Ok(output)
273    }
274
275    fn write(&self, ctx: &OperationContext, path: &str, _: OpWrite) -> Result<Self::Writer> {
276        let output: D1Writer = {
277            let p = build_abs_path(&self.root, path);
278            Ok(D1Writer::new(self.core.clone(), ctx.clone(), p))
279        }?;
280
281        Ok(output)
282    }
283
284    fn delete(&self, ctx: &OperationContext) -> Result<Self::Deleter> {
285        let output: oio::OneShotDeleter<D1Deleter> = {
286            Ok(oio::OneShotDeleter::new(D1Deleter::new(
287                self.core.clone(),
288                ctx.clone(),
289                self.root.clone(),
290            )))
291        }?;
292
293        Ok(output)
294    }
295
296    fn list(&self, _ctx: &OperationContext, _path: &str, _args: OpList) -> Result<Self::Lister> {
297        Err(Error::new(
298            ErrorKind::Unsupported,
299            "operation is not supported",
300        ))
301    }
302
303    fn copy(
304        &self,
305        _ctx: &OperationContext,
306        _from: &str,
307        _to: &str,
308        _args: OpCopy,
309    ) -> Result<Self::Copier> {
310        Err(Error::new(
311            ErrorKind::Unsupported,
312            "operation is not supported",
313        ))
314    }
315
316    async fn rename(
317        &self,
318        _ctx: &OperationContext,
319        _from: &str,
320        _to: &str,
321        _args: OpRename,
322    ) -> Result<RpRename> {
323        Err(Error::new(
324            ErrorKind::Unsupported,
325            "operation is not supported",
326        ))
327    }
328
329    async fn presign(
330        &self,
331        _ctx: &OperationContext,
332        _path: &str,
333        _args: OpPresign,
334    ) -> Result<RpPresign> {
335        Err(Error::new(
336            ErrorKind::Unsupported,
337            "operation is not supported",
338        ))
339    }
340}
opendal_service_d1/backend.rs

opendal_service_d1/
backend.rs
