opendal_core/blocking/operator.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::time::Duration;
19
20use tokio::runtime::Handle;
21
22use crate::Operator as AsyncOperator;
23use crate::raw::PresignedRequest;
24use crate::types::IntoOperatorUri;
25use crate::*;
26
27/// Use OpenDAL in blocking context.
28///
29/// # Notes
30///
31/// blocking::Operator is a wrapper around [`AsyncOperator`]. It calls async runtimes' `block_on` API to spawn blocking tasks.
32/// Please avoid using blocking::Operator in async context.
33///
34/// # Examples
35///
36/// ## Init in async context
37///
38/// blocking::Operator will use current async context's runtime to handle the async calls.
39///
40/// This is just for initialization. You must use `blocking::Operator` in blocking context.
41///
42/// ```rust,no_run
43/// # use opendal_core::services;
44/// # use opendal_core::blocking;
45/// # use opendal_core::Operator;
46/// # use opendal_core::Result;
47///
48/// #[tokio::main]
49/// async fn main() -> Result<()> {
50/// // Create fs backend builder.
51/// let builder = services::Memory::default();
52/// let op = Operator::new(builder)?;
53///
54/// // Build an `blocking::Operator` with blocking layer to start operating the storage.
55/// let _: blocking::Operator = blocking::Operator::new(op)?;
56///
57/// Ok(())
58/// }
59/// ```
60///
61/// ## In async context with blocking functions
62///
63/// If `blocking::Operator` is called in blocking function, please fetch a [`tokio::runtime::EnterGuard`]
64/// first. You can use [`Handle::try_current`] first to get the handle and then call [`Handle::enter`].
65/// This often happens in the case that async function calls blocking function.
66///
67/// ```rust,no_run
68/// # use opendal_core::services;
69/// # use opendal_core::blocking;
70/// # use opendal_core::Operator;
71/// # use opendal_core::Result;
72///
73/// #[tokio::main]
74/// async fn main() -> Result<()> {
75/// let _ = blocking_fn()?;
76/// Ok(())
77/// }
78///
79/// fn blocking_fn() -> Result<blocking::Operator> {
80/// // Create fs backend builder.
81/// let builder = services::Memory::default();
82/// let op = Operator::new(builder)?;
83///
84/// let handle = tokio::runtime::Handle::try_current().unwrap();
85/// let _guard = handle.enter();
86/// // Build an `blocking::Operator` to start operating the storage.
87/// let op: blocking::Operator = blocking::Operator::new(op)?;
88/// Ok(op)
89/// }
90/// ```
91///
92/// ## In blocking context
93///
94/// In a pure blocking context, we can create a runtime and use it to create the `blocking::Operator`.
95///
96/// > The following code uses a global statically created runtime as an example, please manage the
97/// > runtime on demand.
98///
99/// ```rust,no_run
100/// # use std::sync::LazyLock;
101/// # use opendal_core::services;
102/// # use opendal_core::blocking;
103/// # use opendal_core::Operator;
104/// # use opendal_core::Result;
105///
106/// static RUNTIME: LazyLock<tokio::runtime::Runtime> = LazyLock::new(|| {
107/// tokio::runtime::Builder::new_multi_thread()
108/// .enable_all()
109/// .build()
110/// .unwrap()
111/// });
112///
113/// fn main() -> Result<()> {
114/// // Create fs backend builder.
115/// let builder = services::Memory::default();
116/// let op = Operator::new(builder)?;
117///
118/// // Fetch the `EnterGuard` from global runtime.
119/// let _guard = RUNTIME.enter();
120/// // Build an `blocking::Operator` with blocking layer to start operating the storage.
121/// let _: blocking::Operator = blocking::Operator::new(op)?;
122///
123/// Ok(())
124/// }
125/// ```
126#[derive(Clone, Debug)]
127pub struct Operator {
128 handle: tokio::runtime::Handle,
129 op: AsyncOperator,
130}
131
132impl Operator {
133 /// Create a new `BlockingLayer` with the current runtime's handle
134 pub fn new(op: AsyncOperator) -> Result<Self> {
135 Ok(Self {
136 handle: Handle::try_current()
137 .map_err(|_| Error::new(ErrorKind::Unexpected, "failed to get current handle"))?,
138 op,
139 })
140 }
141
142 /// Spawn a future onto the runtime's worker pool and block until it
143 /// completes.
144 ///
145 /// Unlike [`Handle::block_on`] which polls the future on the **calling**
146 /// thread's stack, this method runs the future on a tokio worker thread
147 /// (typically 8 MB stack) and only uses the calling thread to wait for
148 /// the result. This avoids stack overflows when the async state machine
149 /// is deeply nested (e.g. HF/XET uploads driven from a JVM thread with
150 /// a 1 MB default stack).
151 fn spawn_block<F>(&self, f: F) -> Result<F::Output>
152 where
153 F: std::future::Future + Send + 'static,
154 F::Output: Send + 'static,
155 {
156 self.handle.block_on(self.handle.spawn(f)).map_err(|err| {
157 Error::new(ErrorKind::Unexpected, "blocking task failed").set_source(err)
158 })
159 }
160
161 /// Create a blocking operator from URI based configuration.
162 pub fn from_uri(uri: impl IntoOperatorUri) -> Result<Self> {
163 let op = AsyncOperator::from_uri(uri)?;
164 Self::new(op)
165 }
166
167 /// Get information of underlying accessor.
168 ///
169 /// # Examples
170 ///
171 /// ```
172 /// # use std::sync::Arc;
173 /// use opendal_core::blocking;
174 /// # use anyhow::Result;
175 /// use opendal_core::blocking::Operator;
176 ///
177 /// # fn test(op: blocking::Operator) -> Result<()> {
178 /// let info = op.info();
179 /// # Ok(())
180 /// # }
181 /// ```
182 pub fn info(&self) -> OperatorInfo {
183 self.op.info()
184 }
185}
186
187/// # Operator blocking API.
188impl Operator {
189 /// Create a presigned request for stat.
190 ///
191 /// See [`Operator::presign_stat`] for more details.
192 pub fn presign_stat(&self, path: &str, expire: Duration) -> Result<PresignedRequest> {
193 self.handle.block_on(self.op.presign_stat(path, expire))
194 }
195
196 /// Create a presigned request for stat with additional options.
197 pub fn presign_stat_options(
198 &self,
199 path: &str,
200 expire: Duration,
201 opts: options::StatOptions,
202 ) -> Result<PresignedRequest> {
203 let op = self.op.clone();
204 let path = path.to_string();
205 self.spawn_block(async move { op.presign_stat_options(&path, expire, opts).await })?
206 }
207
208 /// Create a presigned request for read.
209 ///
210 /// See [`Operator::presign_read`] for more details.
211 pub fn presign_read(&self, path: &str, expire: Duration) -> Result<PresignedRequest> {
212 self.handle.block_on(self.op.presign_read(path, expire))
213 }
214
215 /// Create a presigned request for read with additional options.
216 pub fn presign_read_options(
217 &self,
218 path: &str,
219 expire: Duration,
220 opts: options::ReadOptions,
221 ) -> Result<PresignedRequest> {
222 let op = self.op.clone();
223 let path = path.to_string();
224 self.spawn_block(async move { op.presign_read_options(&path, expire, opts).await })?
225 }
226
227 /// Create a presigned request for write.
228 ///
229 /// See [`Operator::presign_write`] for more details.
230 pub fn presign_write(&self, path: &str, expire: Duration) -> Result<PresignedRequest> {
231 self.handle.block_on(self.op.presign_write(path, expire))
232 }
233
234 /// Create a presigned request for write with additional options.
235 pub fn presign_write_options(
236 &self,
237 path: &str,
238 expire: Duration,
239 opts: options::WriteOptions,
240 ) -> Result<PresignedRequest> {
241 let op = self.op.clone();
242 let path = path.to_string();
243 self.spawn_block(async move { op.presign_write_options(&path, expire, opts).await })?
244 }
245
246 /// Create a presigned request for delete.
247 ///
248 /// See [`Operator::presign_delete`] for more details.
249 pub fn presign_delete(&self, path: &str, expire: Duration) -> Result<PresignedRequest> {
250 self.handle.block_on(self.op.presign_delete(path, expire))
251 }
252
253 /// Create a presigned request for delete with additional options.
254 pub fn presign_delete_options(
255 &self,
256 path: &str,
257 expire: Duration,
258 opts: options::DeleteOptions,
259 ) -> Result<PresignedRequest> {
260 let op = self.op.clone();
261 let path = path.to_string();
262 self.spawn_block(async move { op.presign_delete_options(&path, expire, opts).await })?
263 }
264
265 /// Get given path's metadata.
266 ///
267 /// # Behavior
268 ///
269 /// ## Services that support `create_dir`
270 ///
271 /// `test` and `test/` may vary in some services such as S3. However, on a local file system,
272 /// they're identical. Therefore, the behavior of `stat("test")` and `stat("test/")` might differ
273 /// in certain edge cases. Always use `stat("test/")` when you need to access a directory if possible.
274 ///
275 /// Here are the behavior list:
276 ///
277 /// | Case | Path | Result |
278 /// |------------------------|-----------------|--------------------------------------------|
279 /// | stat existing dir | `abc/` | Metadata with dir mode |
280 /// | stat existing file | `abc/def_file` | Metadata with file mode |
281 /// | stat dir without `/` | `abc/def_dir` | Error `NotFound` or metadata with dir mode |
282 /// | stat file with `/` | `abc/def_file/` | Error `NotFound` |
283 /// | stat not existing path | `xyz` | Error `NotFound` |
284 ///
285 /// Refer to [RFC: List Prefix][crate::docs::rfcs::rfc_3243_list_prefix] for more details.
286 ///
287 /// ## Services that not support `create_dir`
288 ///
289 /// For services that not support `create_dir`, `stat("test/")` will return `NotFound` even
290 /// when `test/abc` exists since the service won't have the concept of dir. There is nothing
291 /// we can do about this.
292 ///
293 /// # Examples
294 ///
295 /// ## Check if file exists
296 ///
297 /// ```
298 /// # use anyhow::Result;
299 /// # use futures::io;
300 /// use opendal_core::blocking;
301 /// # use opendal_core::blocking::Operator;
302 /// use opendal_core::ErrorKind;
303 /// #
304 /// # fn test(op: blocking::Operator) -> Result<()> {
305 /// if let Err(e) = op.stat("test") {
306 /// if e.kind() == ErrorKind::NotFound {
307 /// println!("file not exist")
308 /// }
309 /// }
310 /// # Ok(())
311 /// # }
312 /// ```
313 pub fn stat(&self, path: &str) -> Result<Metadata> {
314 self.stat_options(path, options::StatOptions::default())
315 }
316
317 /// Get given path's metadata with extra options.
318 ///
319 /// # Behavior
320 ///
321 /// ## Services that support `create_dir`
322 ///
323 /// `test` and `test/` may vary in some services such as S3. However, on a local file system,
324 /// they're identical. Therefore, the behavior of `stat("test")` and `stat("test/")` might differ
325 /// in certain edge cases. Always use `stat("test/")` when you need to access a directory if possible.
326 ///
327 /// Here are the behavior list:
328 ///
329 /// | Case | Path | Result |
330 /// |------------------------|-----------------|--------------------------------------------|
331 /// | stat existing dir | `abc/` | Metadata with dir mode |
332 /// | stat existing file | `abc/def_file` | Metadata with file mode |
333 /// | stat dir without `/` | `abc/def_dir` | Error `NotFound` or metadata with dir mode |
334 /// | stat file with `/` | `abc/def_file/` | Error `NotFound` |
335 /// | stat not existing path | `xyz` | Error `NotFound` |
336 ///
337 /// Refer to [RFC: List Prefix][crate::docs::rfcs::rfc_3243_list_prefix] for more details.
338 ///
339 /// ## Services that not support `create_dir`
340 ///
341 /// For services that not support `create_dir`, `stat("test/")` will return `NotFound` even
342 /// when `test/abc` exists since the service won't have the concept of dir. There is nothing
343 /// we can do about this.
344 pub fn stat_options(&self, path: &str, opts: options::StatOptions) -> Result<Metadata> {
345 let op = self.op.clone();
346 let path = path.to_string();
347 self.spawn_block(async move { op.stat_options(&path, opts).await })?
348 }
349
350 /// Check if this path exists or not.
351 ///
352 /// # Example
353 ///
354 /// ```no_run
355 /// use anyhow::Result;
356 /// use opendal_core::blocking;
357 /// use opendal_core::blocking::Operator;
358 /// fn test(op: blocking::Operator) -> Result<()> {
359 /// let _ = op.exists("test")?;
360 ///
361 /// Ok(())
362 /// }
363 /// ```
364 pub fn exists(&self, path: &str) -> Result<bool> {
365 let r = self.stat(path);
366 match r {
367 Ok(_) => Ok(true),
368 Err(err) => match err.kind() {
369 ErrorKind::NotFound => Ok(false),
370 _ => Err(err),
371 },
372 }
373 }
374
375 /// Create a dir at given path.
376 ///
377 /// # Notes
378 ///
379 /// To indicate that a path is a directory, it is compulsory to include
380 /// a trailing / in the path. Failure to do so may result in
381 /// `NotADirectory` error being returned by OpenDAL.
382 ///
383 /// # Behavior
384 ///
385 /// - Create on existing dir will succeed.
386 /// - Create dir is always recursive, works like `mkdir -p`
387 ///
388 /// # Examples
389 ///
390 /// ```no_run
391 /// # use opendal_core::Result;
392 /// use opendal_core::blocking;
393 /// # use opendal_core::blocking::Operator;
394 /// # use futures::TryStreamExt;
395 /// # fn test(op: blocking::Operator) -> Result<()> {
396 /// op.create_dir("path/to/dir/")?;
397 /// # Ok(())
398 /// # }
399 /// ```
400 pub fn create_dir(&self, path: &str) -> Result<()> {
401 let op = self.op.clone();
402 let path = path.to_string();
403 self.spawn_block(async move { op.create_dir(&path).await })?
404 }
405
406 /// Read the whole path into a bytes.
407 ///
408 /// This function will allocate a new bytes internally. For more precise memory control or
409 /// reading data lazily, please use [`blocking::Operator::reader`]
410 ///
411 /// # Examples
412 ///
413 /// ```no_run
414 /// # use opendal_core::Result;
415 /// use opendal_core::blocking;
416 /// # use opendal_core::blocking::Operator;
417 /// #
418 /// # fn test(op: blocking::Operator) -> Result<()> {
419 /// let bs = op.read("path/to/file")?;
420 /// # Ok(())
421 /// # }
422 /// ```
423 pub fn read(&self, path: &str) -> Result<Buffer> {
424 self.read_options(path, options::ReadOptions::default())
425 }
426
427 /// Read the whole path into a bytes with extra options.
428 ///
429 /// This function will allocate a new bytes internally. For more precise memory control or
430 /// reading data lazily, please use [`blocking::Operator::reader`]
431 pub fn read_options(&self, path: &str, opts: options::ReadOptions) -> Result<Buffer> {
432 let op = self.op.clone();
433 let path = path.to_string();
434 self.spawn_block(async move { op.read_options(&path, opts).await })?
435 }
436
437 /// Create a new reader which can read the whole path.
438 ///
439 /// # Examples
440 ///
441 /// ```no_run
442 /// # use opendal_core::Result;
443 /// use opendal_core::blocking;
444 /// # use opendal_core::blocking::Operator;
445 /// # use futures::TryStreamExt;
446 /// # fn test(op: blocking::Operator) -> Result<()> {
447 /// let r = op.reader("path/to/file")?;
448 /// # Ok(())
449 /// # }
450 /// ```
451 pub fn reader(&self, path: &str) -> Result<blocking::Reader> {
452 self.reader_options(path, options::ReaderOptions::default())
453 }
454
455 /// Create a new reader with extra options
456 pub fn reader_options(
457 &self,
458 path: &str,
459 opts: options::ReaderOptions,
460 ) -> Result<blocking::Reader> {
461 let r = self.handle.block_on(self.op.reader_options(path, opts))?;
462 Ok(blocking::Reader::new(self.handle.clone(), r))
463 }
464
465 /// Write bytes into given path.
466 ///
467 /// # Notes
468 ///
469 /// - Write will make sure all bytes has been written, or an error will be returned.
470 ///
471 /// # Examples
472 ///
473 /// ```no_run
474 /// # use opendal_core::Result;
475 /// # use opendal_core::blocking::Operator;
476 /// # use futures::StreamExt;
477 /// # use futures::SinkExt;
478 /// use bytes::Bytes;
479 /// use opendal_core::blocking;
480 ///
481 /// # fn test(op: blocking::Operator) -> Result<()> {
482 /// op.write("path/to/file", vec![0; 4096])?;
483 /// # Ok(())
484 /// # }
485 /// ```
486 pub fn write(&self, path: &str, bs: impl Into<Buffer>) -> Result<Metadata> {
487 self.write_options(path, bs, options::WriteOptions::default())
488 }
489
490 /// Write data with options.
491 ///
492 /// # Notes
493 ///
494 /// - Write will make sure all bytes has been written, or an error will be returned.
495 pub fn write_options(
496 &self,
497 path: &str,
498 bs: impl Into<Buffer>,
499 opts: options::WriteOptions,
500 ) -> Result<Metadata> {
501 let op = self.op.clone();
502 let path = path.to_string();
503 let bs = bs.into();
504 self.spawn_block(async move { op.write_options(&path, bs, opts).await })?
505 }
506
507 /// Write multiple bytes into given path.
508 ///
509 /// # Notes
510 ///
511 /// - Write will make sure all bytes has been written, or an error will be returned.
512 ///
513 /// # Examples
514 ///
515 /// ```no_run
516 /// # use opendal_core::Result;
517 /// # use opendal_core::blocking;
518 /// # use opendal_core::blocking::Operator;
519 /// # use futures::StreamExt;
520 /// # use futures::SinkExt;
521 /// use bytes::Bytes;
522 ///
523 /// # fn test(op: blocking::Operator) -> Result<()> {
524 /// let mut w = op.writer("path/to/file")?;
525 /// w.write(vec![0; 4096])?;
526 /// w.write(vec![1; 4096])?;
527 /// w.close()?;
528 /// # Ok(())
529 /// # }
530 /// ```
531 pub fn writer(&self, path: &str) -> Result<blocking::Writer> {
532 self.writer_options(path, options::WriteOptions::default())
533 }
534
535 /// Create a new writer with extra options
536 pub fn writer_options(
537 &self,
538 path: &str,
539 opts: options::WriteOptions,
540 ) -> Result<blocking::Writer> {
541 let w = self.handle.block_on(self.op.writer_options(path, opts))?;
542 Ok(blocking::Writer::new(self.handle.clone(), w))
543 }
544
545 /// Copy a file from `from` to `to`.
546 ///
547 /// # Notes
548 ///
549 /// - `from` and `to` must be a file.
550 /// - `to` will be overwritten if it exists.
551 /// - If `from` and `to` are the same, nothing will happen.
552 /// - `copy` is idempotent. For same `from` and `to` input, the result will be the same.
553 ///
554 /// # Examples
555 ///
556 /// ```
557 /// # use opendal_core::Result;
558 /// use opendal_core::blocking;
559 /// # use opendal_core::blocking::Operator;
560 ///
561 /// # fn test(op: blocking::Operator) -> Result<()> {
562 /// op.copy("path/to/file", "path/to/file2")?;
563 /// # Ok(())
564 /// # }
565 /// ```
566 pub fn copy(&self, from: &str, to: &str) -> Result<Metadata> {
567 self.copy_options(from, to, options::CopyOptions::default())
568 }
569
570 /// Copy a file from `from` to `to` with additional options.
571 pub fn copy_options(
572 &self,
573 from: &str,
574 to: &str,
575 opts: options::CopyOptions,
576 ) -> Result<Metadata> {
577 let op = self.op.clone();
578 let from = from.to_string();
579 let to = to.to_string();
580 self.spawn_block(async move { op.copy_options(&from, &to, opts).await })?
581 }
582
583 /// Create a copier from `from` to `to`.
584 ///
585 /// This function creates a new [`blocking::Copier`] that implements
586 /// `Iterator<Item = Result<usize>>`.
587 pub fn copier(&self, from: &str, to: &str) -> Result<blocking::Copier> {
588 self.copier_options(from, to, options::CopyOptions::default())
589 }
590
591 /// Create a copier from `from` to `to` with additional options.
592 pub fn copier_options(
593 &self,
594 from: &str,
595 to: &str,
596 opts: options::CopyOptions,
597 ) -> Result<blocking::Copier> {
598 let copier = self
599 .handle
600 .block_on(self.op.copier_options(from, to, opts))?;
601 Ok(blocking::Copier::new(self.handle.clone(), copier))
602 }
603
604 /// Rename a file from `from` to `to`.
605 ///
606 /// # Notes
607 ///
608 /// - `from` and `to` must be a file.
609 /// - `to` will be overwritten if it exists.
610 /// - If `from` and `to` are the same, an `IsSameFile` error will occur.
611 ///
612 /// # Examples
613 ///
614 /// ```
615 /// # use opendal_core::Result;
616 /// use opendal_core::blocking;
617 /// # use opendal_core::blocking::Operator;
618 ///
619 /// # fn test(op: blocking::Operator) -> Result<()> {
620 /// op.rename("path/to/file", "path/to/file2")?;
621 /// # Ok(())
622 /// # }
623 /// ```
624 pub fn rename(&self, from: &str, to: &str) -> Result<()> {
625 self.rename_options(from, to, options::RenameOptions::default())
626 }
627
628 /// Rename a file from `from` to `to` with additional options.
629 ///
630 /// # Options
631 ///
632 /// Visit [`options::RenameOptions`] for all available options.
633 ///
634 /// # Examples
635 ///
636 /// ```
637 /// use opendal_core::blocking;
638 /// use opendal_core::options::RenameOptions;
639 /// use opendal_core::Result;
640 ///
641 /// fn rename_with_options(op: blocking::Operator) -> Result<()> {
642 /// let mut opts = RenameOptions::default();
643 /// opts.if_not_exists = true;
644 /// op.rename_options("path/to/file", "path/to/file2", opts)?;
645 /// Ok(())
646 /// }
647 /// ```
648 pub fn rename_options(&self, from: &str, to: &str, opts: options::RenameOptions) -> Result<()> {
649 let op = self.op.clone();
650 let from = from.to_string();
651 let to = to.to_string();
652 self.spawn_block(async move { op.rename_options(&from, &to, opts).await })?
653 }
654
655 /// Delete given path.
656 ///
657 /// # Notes
658 ///
659 /// - Delete not existing error won't return errors.
660 ///
661 /// # Examples
662 ///
663 /// ```no_run
664 /// # use anyhow::Result;
665 /// # use futures::io;
666 /// use opendal_core::blocking;
667 /// # use opendal_core::blocking::Operator;
668 /// # fn test(op: blocking::Operator) -> Result<()> {
669 /// op.delete("path/to/file")?;
670 /// # Ok(())
671 /// # }
672 /// ```
673 pub fn delete(&self, path: &str) -> Result<()> {
674 self.delete_options(path, options::DeleteOptions::default())
675 }
676
677 /// Delete given path with options.
678 ///
679 /// # Notes
680 ///
681 /// - Delete not existing error won't return errors.
682 pub fn delete_options(&self, path: &str, opts: options::DeleteOptions) -> Result<()> {
683 let op = self.op.clone();
684 let path = path.to_string();
685 self.spawn_block(async move { op.delete_options(&path, opts).await })?
686 }
687
688 /// Delete an infallible iterator of paths.
689 ///
690 /// Also see:
691 ///
692 /// - [`blocking::Operator::delete_try_iter`]: delete an fallible iterator of paths.
693 pub fn delete_iter<I, D>(&self, iter: I) -> Result<()>
694 where
695 I: IntoIterator<Item = D>,
696 D: IntoDeleteInput,
697 {
698 self.handle.block_on(self.op.delete_iter(iter))
699 }
700
701 /// Delete a fallible iterator of paths.
702 ///
703 /// Also see:
704 ///
705 /// - [`blocking::Operator::delete_iter`]: delete an infallible iterator of paths.
706 pub fn delete_try_iter<I, D>(&self, try_iter: I) -> Result<()>
707 where
708 I: IntoIterator<Item = Result<D>>,
709 D: IntoDeleteInput,
710 {
711 self.handle.block_on(self.op.delete_try_iter(try_iter))
712 }
713
714 /// Create a [`BlockingDeleter`] to continuously remove content from storage.
715 ///
716 /// It leverages batch deletion capabilities provided by storage services for efficient removal.
717 ///
718 /// Users can have more control over the deletion process by using [`BlockingDeleter`] directly.
719 pub fn deleter(&self) -> Result<blocking::Deleter> {
720 blocking::Deleter::create(
721 self.handle.clone(),
722 self.handle.block_on(self.op.deleter())?,
723 )
724 }
725
726 /// Remove the path and all nested dirs and files recursively.
727 ///
728 /// # Deprecated
729 ///
730 /// This method is deprecated since v0.55.0. Use [`blocking::Operator::delete_options`] with
731 /// `recursive: true` instead.
732 ///
733 /// ## Migration Example
734 ///
735 /// Instead of:
736 /// ```ignore
737 /// op.remove_all("path/to/dir")?;
738 /// ```
739 ///
740 /// Use:
741 /// ```ignore
742 /// use opendal_core::options::DeleteOptions;
743 /// op.delete_options("path/to/dir", DeleteOptions {
744 /// recursive: true,
745 /// ..Default::default()
746 /// })?;
747 /// ```
748 ///
749 /// # Notes
750 ///
751 /// If underlying services support delete in batch, we will use batch
752 /// delete instead.
753 ///
754 /// # Examples
755 ///
756 /// ```
757 /// # use anyhow::Result;
758 /// # use futures::io;
759 /// use opendal_core::blocking;
760 /// # use opendal_core::blocking::Operator;
761 /// # fn test(op: blocking::Operator) -> Result<()> {
762 /// op.remove_all("path/to/dir")?;
763 /// # Ok(())
764 /// # }
765 /// ```
766 #[deprecated(
767 since = "0.55.0",
768 note = "Use `delete_options` with `recursive: true` instead"
769 )]
770 #[allow(deprecated)]
771 pub fn remove_all(&self, path: &str) -> Result<()> {
772 self.delete_options(
773 path,
774 options::DeleteOptions {
775 recursive: true,
776 ..Default::default()
777 },
778 )
779 }
780
781 /// List entries whose paths start with the given prefix `path`.
782 ///
783 /// # Semantics
784 ///
785 /// - Listing is **prefix-based**; it doesn't require the parent directory to exist.
786 /// - If `path` itself exists, it is returned as an entry along with prefixed children.
787 /// - If `path` is missing but deeper objects exist, the list succeeds and returns those prefixed entries instead of an error.
788 /// - Set `recursive` in [`options::ListOptions`] via [`list_options`](Self::list_options) to walk all descendants; the default returns only immediate children when delimiter is supported.
789 ///
790 /// ## Streaming List
791 ///
792 /// This function materializes the full result in memory. For large listings, prefer [`blocking::Operator::lister`] to stream entries.
793 ///
794 /// # Examples
795 ///
796 /// ```no_run
797 /// # use anyhow::Result;
798 /// use opendal_core::blocking;
799 /// use opendal_core::blocking::Operator;
800 /// use opendal_core::EntryMode;
801 /// # fn test(op: blocking::Operator) -> Result<()> {
802 /// let mut entries = op.list("path/to/dir/")?;
803 /// for entry in entries {
804 /// match entry.metadata().mode() {
805 /// EntryMode::FILE => {
806 /// println!("Handling file")
807 /// }
808 /// EntryMode::DIR => {
809 /// println!("Handling dir {}", entry.path())
810 /// }
811 /// EntryMode::Unknown => continue,
812 /// }
813 /// }
814 /// # Ok(())
815 /// # }
816 /// ```
817 pub fn list(&self, path: &str) -> Result<Vec<Entry>> {
818 self.list_options(path, options::ListOptions::default())
819 }
820
821 /// List entries whose paths start with the given prefix `path` with additional options.
822 ///
823 /// # Semantics
824 ///
825 /// Inherits the prefix semantics described in [`Operator::list`] (blocking variant). It returns `path` itself if present and tolerates missing parents when prefixed objects exist.
826 ///
827 /// ## Streaming List
828 ///
829 /// This function materializes the full result in memory. For large listings, prefer [`blocking::Operator::lister`] to stream entries.
830 ///
831 /// ## Options
832 ///
833 /// See [`options::ListOptions`] for the full set. Common knobs: traversal (`recursive`), pagination (`limit`, `start_after`), and versioning (`versions`, `deleted`).
834 pub fn list_options(&self, path: &str, opts: options::ListOptions) -> Result<Vec<Entry>> {
835 let op = self.op.clone();
836 let path = path.to_string();
837 self.spawn_block(async move { op.list_options(&path, opts).await })?
838 }
839
840 /// Create a streaming lister for entries whose paths start with the given prefix `path`.
841 ///
842 /// This function creates a new [`BlockingLister`]; dropping it stops listing.
843 ///
844 /// # Semantics
845 ///
846 /// Shares the same prefix semantics as [`blocking::Operator::list`]: parent directory is optional; `path` itself is yielded if present; missing parents with deeper objects are accepted.
847 ///
848 /// ## Options
849 ///
850 /// Accepts the same [`options::ListOptions`] as [`list_options`](Self::list_options): traversal (`recursive`), pagination (`limit`, `start_after`), and versioning (`versions`, `deleted`).
851 ///
852 /// # Examples
853 ///
854 /// ```no_run
855 /// # use anyhow::Result;
856 /// # use futures::io;
857 /// use futures::TryStreamExt;
858 /// use opendal_core::blocking;
859 /// use opendal_core::blocking::Operator;
860 /// use opendal_core::EntryMode;
861 /// # fn test(op: blocking::Operator) -> Result<()> {
862 /// let mut ds = op.lister("path/to/dir/")?;
863 /// for de in ds {
864 /// let de = de?;
865 /// match de.metadata().mode() {
866 /// EntryMode::FILE => {
867 /// println!("Handling file")
868 /// }
869 /// EntryMode::DIR => {
870 /// println!("Handling dir like start a new list via meta.path()")
871 /// }
872 /// EntryMode::Unknown => continue,
873 /// }
874 /// }
875 /// # Ok(())
876 /// # }
877 /// ```
878 pub fn lister(&self, path: &str) -> Result<blocking::Lister> {
879 self.lister_options(path, options::ListOptions::default())
880 }
881
882 /// List entries under a prefix as an iterator with options.
883 ///
884 /// This function creates a new handle to stream entries and inherits the prefix semantics of [`blocking::Operator::list`].
885 ///
886 /// ## Options
887 ///
888 /// Same as [`lister`](Self::lister); see [`options::ListOptions`] for traversal, pagination, and versioning knobs.
889 pub fn lister_options(
890 &self,
891 path: &str,
892 opts: options::ListOptions,
893 ) -> Result<blocking::Lister> {
894 let l = self.handle.block_on(self.op.lister_options(path, opts))?;
895 Ok(blocking::Lister::new(self.handle.clone(), l))
896 }
897
898 /// Check if this operator can work correctly.
899 ///
900 /// We will send a `list` request to path and return any errors we met.
901 ///
902 /// ```
903 /// # use std::sync::Arc;
904 /// # use anyhow::Result;
905 /// use opendal_core::blocking;
906 /// use opendal_core::blocking::Operator;
907 /// use opendal_core::ErrorKind;
908 ///
909 /// # fn test(op: blocking::Operator) -> Result<()> {
910 /// op.check()?;
911 /// # Ok(())
912 /// # }
913 /// ```
914 pub fn check(&self) -> Result<()> {
915 let mut ds = self.lister("/")?;
916
917 match ds.next() {
918 Some(Err(e)) if e.kind() != ErrorKind::NotFound => Err(e),
919 _ => Ok(()),
920 }
921 }
922}
923
924impl From<Operator> for AsyncOperator {
925 fn from(val: Operator) -> Self {
926 val.op
927 }
928}