opendal/services/koofr/
backend.rs1use std::fmt::Debug;
19use std::sync::Arc;
20
21use bytes::Buf;
22use http::Response;
23use http::StatusCode;
24use log::debug;
25use mea::mutex::Mutex;
26use mea::once::OnceCell;
27
28use super::KOOFR_SCHEME;
29use super::config::KoofrConfig;
30use super::core::File;
31use super::core::KoofrCore;
32use super::core::KoofrSigner;
33use super::deleter::KoofrDeleter;
34use super::error::parse_error;
35use super::lister::KoofrLister;
36use super::writer::KoofrWriter;
37use super::writer::KoofrWriters;
38use crate::raw::*;
39use crate::*;
40
41#[doc = include_str!("docs.md")]
43#[derive(Default)]
44pub struct KoofrBuilder {
45 pub(super) config: KoofrConfig,
46
47 #[deprecated(since = "0.53.0", note = "Use `Operator::update_http_client` instead")]
48 pub(super) http_client: Option<HttpClient>,
49}
50
51impl Debug for KoofrBuilder {
52 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
53 f.debug_struct("KoofrBuilder")
54 .field("config", &self.config)
55 .finish_non_exhaustive()
56 }
57}
58
59impl KoofrBuilder {
60 pub fn root(mut self, root: &str) -> Self {
64 self.config.root = if root.is_empty() {
65 None
66 } else {
67 Some(root.to_string())
68 };
69
70 self
71 }
72
73 pub fn endpoint(mut self, endpoint: &str) -> Self {
77 self.config.endpoint = endpoint.to_string();
78
79 self
80 }
81
82 pub fn email(mut self, email: &str) -> Self {
86 self.config.email = email.to_string();
87
88 self
89 }
90
91 pub fn password(mut self, password: &str) -> Self {
102 self.config.password = if password.is_empty() {
103 None
104 } else {
105 Some(password.to_string())
106 };
107
108 self
109 }
110
111 #[deprecated(since = "0.53.0", note = "Use `Operator::update_http_client` instead")]
118 #[allow(deprecated)]
119 pub fn http_client(mut self, client: HttpClient) -> Self {
120 self.http_client = Some(client);
121 self
122 }
123}
124
125impl Builder for KoofrBuilder {
126 type Config = KoofrConfig;
127
128 fn build(self) -> Result<impl Access> {
130 debug!("backend build started: {:?}", &self);
131
132 let root = normalize_root(&self.config.root.clone().unwrap_or_default());
133 debug!("backend use root {}", &root);
134
135 if self.config.endpoint.is_empty() {
136 return Err(Error::new(ErrorKind::ConfigInvalid, "endpoint is empty")
137 .with_operation("Builder::build")
138 .with_context("service", KOOFR_SCHEME));
139 }
140
141 debug!("backend use endpoint {}", &self.config.endpoint);
142
143 if self.config.email.is_empty() {
144 return Err(Error::new(ErrorKind::ConfigInvalid, "email is empty")
145 .with_operation("Builder::build")
146 .with_context("service", KOOFR_SCHEME));
147 }
148
149 debug!("backend use email {}", &self.config.email);
150
151 let password = match &self.config.password {
152 Some(password) => Ok(password.clone()),
153 None => Err(Error::new(ErrorKind::ConfigInvalid, "password is empty")
154 .with_operation("Builder::build")
155 .with_context("service", KOOFR_SCHEME)),
156 }?;
157
158 let signer = Arc::new(Mutex::new(KoofrSigner::default()));
159
160 Ok(KoofrBackend {
161 core: Arc::new(KoofrCore {
162 info: {
163 let am = AccessorInfo::default();
164 am.set_scheme(KOOFR_SCHEME)
165 .set_root(&root)
166 .set_native_capability(Capability {
167 stat: true,
168
169 create_dir: true,
170
171 read: true,
172
173 write: true,
174 write_can_empty: true,
175
176 delete: true,
177
178 rename: true,
179
180 copy: true,
181
182 list: true,
183
184 shared: true,
185
186 ..Default::default()
187 });
188
189 #[allow(deprecated)]
191 if let Some(client) = self.http_client {
192 am.update_http_client(|_| client);
193 }
194
195 am.into()
196 },
197 root,
198 endpoint: self.config.endpoint.clone(),
199 email: self.config.email.clone(),
200 password,
201 mount_id: OnceCell::new(),
202 signer,
203 }),
204 })
205 }
206}
207
208#[derive(Debug, Clone)]
210pub struct KoofrBackend {
211 core: Arc<KoofrCore>,
212}
213
214impl Access for KoofrBackend {
215 type Reader = HttpBody;
216 type Writer = KoofrWriters;
217 type Lister = oio::PageLister<KoofrLister>;
218 type Deleter = oio::OneShotDeleter<KoofrDeleter>;
219
220 fn info(&self) -> Arc<AccessorInfo> {
221 self.core.info.clone()
222 }
223
224 async fn create_dir(&self, path: &str, _: OpCreateDir) -> Result<RpCreateDir> {
225 self.core.ensure_dir_exists(path).await?;
226 self.core
227 .create_dir(&build_abs_path(&self.core.root, path))
228 .await?;
229 Ok(RpCreateDir::default())
230 }
231
232 async fn stat(&self, path: &str, _args: OpStat) -> Result<RpStat> {
233 let path = build_rooted_abs_path(&self.core.root, path);
234 let resp = self.core.info(&path).await?;
235
236 let status = resp.status();
237
238 match status {
239 StatusCode::OK => {
240 let bs = resp.into_body();
241
242 let file: File =
243 serde_json::from_reader(bs.reader()).map_err(new_json_deserialize_error)?;
244
245 let mode = if file.ty == "dir" {
246 EntryMode::DIR
247 } else {
248 EntryMode::FILE
249 };
250
251 let mut md = Metadata::new(mode);
252
253 md.set_content_length(file.size)
254 .set_content_type(&file.content_type)
255 .set_last_modified(Timestamp::from_millisecond(file.modified)?);
256
257 Ok(RpStat::new(md))
258 }
259 _ => Err(parse_error(resp)),
260 }
261 }
262
263 async fn read(&self, path: &str, args: OpRead) -> Result<(RpRead, Self::Reader)> {
264 let resp = self.core.get(path, args.range()).await?;
265
266 let status = resp.status();
267 match status {
268 StatusCode::OK | StatusCode::PARTIAL_CONTENT => {
269 Ok((RpRead::default(), resp.into_body()))
270 }
271 _ => {
272 let (part, mut body) = resp.into_parts();
273 let buf = body.to_buffer().await?;
274 Err(parse_error(Response::from_parts(part, buf)))
275 }
276 }
277 }
278
279 async fn write(&self, path: &str, _args: OpWrite) -> Result<(RpWrite, Self::Writer)> {
280 let writer = KoofrWriter::new(self.core.clone(), path.to_string());
281
282 let w = oio::OneShotWriter::new(writer);
283
284 Ok((RpWrite::default(), w))
285 }
286
287 async fn delete(&self) -> Result<(RpDelete, Self::Deleter)> {
288 Ok((
289 RpDelete::default(),
290 oio::OneShotDeleter::new(KoofrDeleter::new(self.core.clone())),
291 ))
292 }
293
294 async fn list(&self, path: &str, _args: OpList) -> Result<(RpList, Self::Lister)> {
295 let l = KoofrLister::new(self.core.clone(), path);
296 Ok((RpList::default(), oio::PageLister::new(l)))
297 }
298
299 async fn copy(&self, from: &str, to: &str, _args: OpCopy) -> Result<RpCopy> {
300 self.core.ensure_dir_exists(to).await?;
301
302 if from == to {
303 return Ok(RpCopy::default());
304 }
305
306 let resp = self.core.remove(to).await?;
307
308 let status = resp.status();
309
310 if status != StatusCode::OK && status != StatusCode::NOT_FOUND {
311 return Err(parse_error(resp));
312 }
313
314 let resp = self.core.copy(from, to).await?;
315
316 let status = resp.status();
317
318 match status {
319 StatusCode::OK => Ok(RpCopy::default()),
320 _ => Err(parse_error(resp)),
321 }
322 }
323
324 async fn rename(&self, from: &str, to: &str, _args: OpRename) -> Result<RpRename> {
325 self.core.ensure_dir_exists(to).await?;
326
327 if from == to {
328 return Ok(RpRename::default());
329 }
330
331 let resp = self.core.remove(to).await?;
332
333 let status = resp.status();
334
335 if status != StatusCode::OK && status != StatusCode::NOT_FOUND {
336 return Err(parse_error(resp));
337 }
338
339 let resp = self.core.move_object(from, to).await?;
340
341 let status = resp.status();
342
343 match status {
344 StatusCode::OK => Ok(RpRename::default()),
345 _ => Err(parse_error(resp)),
346 }
347 }
348}