opendal/services/mongodb/
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::sync::Arc;
19
20use tokio::sync::OnceCell;
21
22use super::MONGODB_SCHEME;
23use super::config::MongodbConfig;
24use super::core::*;
25use super::deleter::MongodbDeleter;
26use super::writer::MongodbWriter;
27use crate::raw::*;
28use crate::*;
29
30#[doc = include_str!("docs.md")]
31#[derive(Debug, Default)]
32pub struct MongodbBuilder {
33    pub(super) config: MongodbConfig,
34}
35
36impl MongodbBuilder {
37    /// Set the connection_string of the MongoDB service.
38    ///
39    /// This connection string is used to connect to the MongoDB service. It typically follows the format:
40    ///
41    /// ## Format
42    ///
43    /// `mongodb://[username:password@]host1[:port1][,...hostN[:portN]][/[defaultauthdb][?options]]`
44    ///
45    /// Examples:
46    ///
47    /// - Connecting to a local MongoDB instance: `mongodb://localhost:27017`
48    /// - Using authentication: `mongodb://myUser:myPassword@localhost:27017/myAuthDB`
49    /// - Specifying authentication mechanism: `mongodb://myUser:myPassword@localhost:27017/myAuthDB?authMechanism=SCRAM-SHA-256`
50    ///
51    /// ## Options
52    ///
53    /// - `authMechanism`: Specifies the authentication method to use. Examples include `SCRAM-SHA-1`, `SCRAM-SHA-256`, and `MONGODB-AWS`.
54    /// - ... (any other options you wish to highlight)
55    ///
56    /// For more information, please refer to [MongoDB Connection String URI Format](https://docs.mongodb.com/manual/reference/connection-string/).
57    pub fn connection_string(mut self, v: &str) -> Self {
58        if !v.is_empty() {
59            self.config.connection_string = Some(v.to_string());
60        }
61        self
62    }
63    /// Set the working directory, all operations will be performed under it.
64    ///
65    /// default: "/"
66    pub fn root(mut self, root: &str) -> Self {
67        self.config.root = if root.is_empty() {
68            None
69        } else {
70            Some(root.to_string())
71        };
72
73        self
74    }
75
76    /// Set the database name of the MongoDB service to read/write.
77    pub fn database(mut self, database: &str) -> Self {
78        if !database.is_empty() {
79            self.config.database = Some(database.to_string());
80        }
81        self
82    }
83
84    /// Set the collection name of the MongoDB service to read/write.
85    pub fn collection(mut self, collection: &str) -> Self {
86        if !collection.is_empty() {
87            self.config.collection = Some(collection.to_string());
88        }
89        self
90    }
91
92    /// Set the key field name of the MongoDB service to read/write.
93    ///
94    /// Default to `key` if not specified.
95    pub fn key_field(mut self, key_field: &str) -> Self {
96        if !key_field.is_empty() {
97            self.config.key_field = Some(key_field.to_string());
98        }
99        self
100    }
101
102    /// Set the value field name of the MongoDB service to read/write.
103    ///
104    /// Default to `value` if not specified.
105    pub fn value_field(mut self, value_field: &str) -> Self {
106        if !value_field.is_empty() {
107            self.config.value_field = Some(value_field.to_string());
108        }
109        self
110    }
111}
112
113impl Builder for MongodbBuilder {
114    type Config = MongodbConfig;
115
116    fn build(self) -> Result<impl Access> {
117        let conn = match &self.config.connection_string.clone() {
118            Some(v) => v.clone(),
119            None => {
120                return Err(
121                    Error::new(ErrorKind::ConfigInvalid, "connection_string is required")
122                        .with_context("service", MONGODB_SCHEME),
123                );
124            }
125        };
126        let database = match &self.config.database.clone() {
127            Some(v) => v.clone(),
128            None => {
129                return Err(Error::new(ErrorKind::ConfigInvalid, "database is required")
130                    .with_context("service", MONGODB_SCHEME));
131            }
132        };
133        let collection = match &self.config.collection.clone() {
134            Some(v) => v.clone(),
135            None => {
136                return Err(
137                    Error::new(ErrorKind::ConfigInvalid, "collection is required")
138                        .with_context("service", MONGODB_SCHEME),
139                );
140            }
141        };
142        let key_field = match &self.config.key_field.clone() {
143            Some(v) => v.clone(),
144            None => "key".to_string(),
145        };
146        let value_field = match &self.config.value_field.clone() {
147            Some(v) => v.clone(),
148            None => "value".to_string(),
149        };
150        let root = normalize_root(
151            self.config
152                .root
153                .clone()
154                .unwrap_or_else(|| "/".to_string())
155                .as_str(),
156        );
157        Ok(MongodbBackend::new(MongodbCore {
158            connection_string: conn,
159            database,
160            collection,
161            collection_instance: OnceCell::new(),
162            key_field,
163            value_field,
164        })
165        .with_normalized_root(root))
166    }
167}
168
169/// Backend for Mongodb services.
170#[derive(Clone, Debug)]
171pub struct MongodbBackend {
172    core: Arc<MongodbCore>,
173    root: String,
174    info: Arc<AccessorInfo>,
175}
176
177impl MongodbBackend {
178    pub fn new(core: MongodbCore) -> Self {
179        let info = AccessorInfo::default();
180        info.set_scheme(MONGODB_SCHEME);
181        info.set_name(&format!("{}/{}", core.database, core.collection));
182        info.set_root("/");
183        info.set_native_capability(Capability {
184            read: true,
185            stat: true,
186            write: true,
187            write_can_empty: true,
188            delete: true,
189            shared: true,
190            ..Default::default()
191        });
192
193        Self {
194            core: Arc::new(core),
195            root: "/".to_string(),
196            info: Arc::new(info),
197        }
198    }
199
200    fn with_normalized_root(mut self, root: String) -> Self {
201        self.info.set_root(&root);
202        self.root = root;
203        self
204    }
205}
206
207impl Access for MongodbBackend {
208    type Reader = Buffer;
209    type Writer = MongodbWriter;
210    type Lister = ();
211    type Deleter = oio::OneShotDeleter<MongodbDeleter>;
212
213    fn info(&self) -> Arc<AccessorInfo> {
214        self.info.clone()
215    }
216
217    async fn stat(&self, path: &str, _: OpStat) -> Result<RpStat> {
218        let p = build_abs_path(&self.root, path);
219
220        if p == build_abs_path(&self.root, "") {
221            Ok(RpStat::new(Metadata::new(EntryMode::DIR)))
222        } else {
223            let bs = self.core.get(&p).await?;
224            match bs {
225                Some(bs) => Ok(RpStat::new(
226                    Metadata::new(EntryMode::FILE).with_content_length(bs.len() as u64),
227                )),
228                None => Err(Error::new(ErrorKind::NotFound, "kv not found in mongodb")),
229            }
230        }
231    }
232
233    async fn read(&self, path: &str, args: OpRead) -> Result<(RpRead, Self::Reader)> {
234        let p = build_abs_path(&self.root, path);
235        let bs = match self.core.get(&p).await? {
236            Some(bs) => bs,
237            None => {
238                return Err(Error::new(ErrorKind::NotFound, "kv not found in mongodb"));
239            }
240        };
241        Ok((RpRead::new(), bs.slice(args.range().to_range_as_usize())))
242    }
243
244    async fn write(&self, path: &str, _: OpWrite) -> Result<(RpWrite, Self::Writer)> {
245        let p = build_abs_path(&self.root, path);
246        Ok((RpWrite::new(), MongodbWriter::new(self.core.clone(), p)))
247    }
248
249    async fn delete(&self) -> Result<(RpDelete, Self::Deleter)> {
250        Ok((
251            RpDelete::default(),
252            oio::OneShotDeleter::new(MongodbDeleter::new(self.core.clone(), self.root.clone())),
253        ))
254    }
255}
opendal/services/mongodb/backend.rs

opendal/services/mongodb/
backend.rs
