You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@opendal.apache.org by xu...@apache.org on 2024/05/17 09:53:06 UTC
(opendal) 01/01: Try bump pyo3
This is an automated email from the ASF dual-hosted git repository.
xuanwo pushed a commit to branch bump-to-0.21
in repository https://gitbox.apache.org/repos/asf/opendal.git
commit effa3322332870ef48966535a3a2404ac7c28953
Author: Xuanwo <gi...@xuanwo.io>
AuthorDate: Fri May 17 17:52:43 2024 +0800
Try bump pyo3
Signed-off-by: Xuanwo <gi...@xuanwo.io>
---
bindings/python/Cargo.toml | 125 ++++++++++---------
bindings/python/src/file.rs | 267 ++++++++++++++++++----------------------
bindings/python/src/lister.rs | 19 ++-
bindings/python/src/operator.rs | 223 ++++++++++++++++-----------------
4 files changed, 297 insertions(+), 337 deletions(-)
diff --git a/bindings/python/Cargo.toml b/bindings/python/Cargo.toml
index f3b4de2c9b..4e03388639 100644
--- a/bindings/python/Cargo.toml
+++ b/bindings/python/Cargo.toml
@@ -29,66 +29,66 @@ version = "0.45.2"
[features]
default = [
- "services-azblob",
- "services-azdls",
- "services-cos",
- "services-fs",
- "services-gcs",
- "services-ghac",
- "services-http",
- "services-ipmfs",
- "services-memory",
- "services-obs",
- "services-oss",
- "services-s3",
- "services-webdav",
- "services-webhdfs",
+ "services-azblob",
+ "services-azdls",
+ "services-cos",
+ "services-fs",
+ "services-gcs",
+ "services-ghac",
+ "services-http",
+ "services-ipmfs",
+ "services-memory",
+ "services-obs",
+ "services-oss",
+ "services-s3",
+ "services-webdav",
+ "services-webhdfs",
]
services-all = [
- "default",
- "services-azfile",
- "services-cacache",
- "services-chainsafe",
- "services-dashmap",
- "services-dropbox",
- # FIXME this service need protoc
- # "services-etcd",
- # FIXME this requires a preinstalled fdb library
- # "services-foundationdb",
- # "services-ftp",
- "services-gdrive",
- # FIXME how to support HDFS services in bindings?
- # "services-hdfs",
- "services-huggingface",
- "services-ipfs",
- "services-memcached",
- "services-mini-moka",
- "services-moka",
- "services-onedrive",
- "services-persy",
- "services-postgresql",
- "services-mysql",
- "services-redb",
- "services-redis",
- # FIXME how to support rocksdb services in bindings?
- # "services-rocksdb",
- "services-sled",
- "services-supabase",
- "services-swift",
- # FIXME this service need protoc
- # "services-tikv",
- "services-vercel-artifacts",
- "services-mongodb",
- "services-gridfs",
- "services-sqlite",
- "services-libsql",
- "services-alluxio",
- "services-b2",
- "services-seafile",
- "services-upyun",
- "services-koofr",
- "services-yandex-disk",
+ "default",
+ "services-azfile",
+ "services-cacache",
+ "services-chainsafe",
+ "services-dashmap",
+ "services-dropbox",
+ # FIXME this service need protoc
+ # "services-etcd",
+ # FIXME this requires a preinstalled fdb library
+ # "services-foundationdb",
+ # "services-ftp",
+ "services-gdrive",
+ # FIXME how to support HDFS services in bindings?
+ # "services-hdfs",
+ "services-huggingface",
+ "services-ipfs",
+ "services-memcached",
+ "services-mini-moka",
+ "services-moka",
+ "services-onedrive",
+ "services-persy",
+ "services-postgresql",
+ "services-mysql",
+ "services-redb",
+ "services-redis",
+ # FIXME how to support rocksdb services in bindings?
+ # "services-rocksdb",
+ "services-sled",
+ "services-supabase",
+ "services-swift",
+ # FIXME this service need protoc
+ # "services-tikv",
+ "services-vercel-artifacts",
+ "services-mongodb",
+ "services-gridfs",
+ "services-sqlite",
+ "services-libsql",
+ "services-alluxio",
+ "services-b2",
+ "services-seafile",
+ "services-upyun",
+ "services-koofr",
+ "services-yandex-disk",
]
# Default services provided by opendal.
@@ -155,17 +155,16 @@ doc = false
bytes = "1.5.0"
futures = "0.3.28"
opendal = { version = "0.46.0", path = "../../core", features = [
- "layers-blocking",
+ "layers-blocking",
] }
-pyo3 = { version = "0.20.1", features = ["abi3", "abi3-py311"] }
-pyo3-asyncio = { version = "0.20", features = ["tokio-runtime"] }
+pyo3 = { version = "0.21.2", features = ["abi3", "abi3-py311", "experimental-async"] }
tokio = "1"
[target.'cfg(unix)'.dependencies.opendal]
features = [
- "layers-blocking",
- # Depend on "openssh" which depends on "tokio-pipe" that is unavailable on Windows.
- "services-sftp",
+ "layers-blocking",
+ # Depend on "openssh" which depends on "tokio-pipe" that is unavailable on Windows.
+ "services-sftp",
]
path = "../../core"
version = "0.46.0"
diff --git a/bindings/python/src/file.rs b/bindings/python/src/file.rs
index b71b903fcb..148f876d74 100644
--- a/bindings/python/src/file.rs
+++ b/bindings/python/src/file.rs
@@ -29,7 +29,6 @@ use futures::{AsyncReadExt, AsyncSeekExt};
use pyo3::buffer::PyBuffer;
use pyo3::exceptions::{PyIOError, PyValueError};
use pyo3::prelude::*;
-use pyo3_asyncio::tokio::future_into_py;
use tokio::sync::Mutex;
use crate::*;
@@ -312,80 +311,76 @@ impl AsyncFile {
#[pymethods]
impl AsyncFile {
/// Read and return at most size bytes, or if size is not given, until EOF.
- pub fn read<'p>(&'p self, py: Python<'p>, size: Option<usize>) -> PyResult<&'p PyAny> {
+ pub async fn read<'p>(&'p self, py: Python<'p>, size: Option<usize>) -> PyResult<Buffer> {
let state = self.0.clone();
- future_into_py(py, async move {
- let mut guard = state.lock().await;
- let reader = match guard.deref_mut() {
- AsyncFileState::Reader(r) => r,
- AsyncFileState::Writer(_) => {
- return Err(PyIOError::new_err(
- "I/O operation failed for reading on write only file.",
- ));
- }
- _ => {
- return Err(PyIOError::new_err(
- "I/O operation failed for reading on closed file.",
- ));
- }
- };
-
- let buffer = match size {
- Some(size) => {
- // TODO: optimize here by using uninit slice.
- let mut bs = vec![0; size];
- let n = reader
- .read(&mut bs)
- .await
- .map_err(|err| PyIOError::new_err(err.to_string()))?;
- bs.truncate(n);
- bs
- }
- None => {
- let mut buffer = Vec::new();
- reader
- .read_to_end(&mut buffer)
- .await
- .map_err(|err| PyIOError::new_err(err.to_string()))?;
- buffer
- }
- };
+ let mut guard = state.lock().await;
+ let reader = match guard.deref_mut() {
+ AsyncFileState::Reader(r) => r,
+ AsyncFileState::Writer(_) => {
+ return Err(PyIOError::new_err(
+ "I/O operation failed for reading on write only file.",
+ ));
+ }
+ _ => {
+ return Err(PyIOError::new_err(
+ "I/O operation failed for reading on closed file.",
+ ));
+ }
+ };
- Python::with_gil(|py| Buffer::new(buffer).into_bytes(py))
- })
+ let buffer = match size {
+ Some(size) => {
+ // TODO: optimize here by using uninit slice.
+ let mut bs = vec![0; size];
+ let n = reader
+ .read(&mut bs)
+ .await
+ .map_err(|err| PyIOError::new_err(err.to_string()))?;
+ bs.truncate(n);
+ bs
+ }
+ None => {
+ let mut buffer = Vec::new();
+ reader
+ .read_to_end(&mut buffer)
+ .await
+ .map_err(|err| PyIOError::new_err(err.to_string()))?;
+ buffer
+ }
+ };
+
+ Ok(Buffer::new(buffer))
}
/// Write bytes into the file.
- pub fn write<'p>(&'p mut self, py: Python<'p>, bs: &'p [u8]) -> PyResult<&'p PyAny> {
+ pub async fn write<'p>(&'p mut self, py: Python<'p>, bs: &'p [u8]) -> PyResult<usize> {
let state = self.0.clone();
// FIXME: can we avoid this clone?
let bs = bs.to_vec();
- future_into_py(py, async move {
- let mut guard = state.lock().await;
- let writer = match guard.deref_mut() {
- AsyncFileState::Reader(_) => {
- return Err(PyIOError::new_err(
- "I/O operation failed for reading on read only file.",
- ));
- }
- AsyncFileState::Writer(w) => w,
- _ => {
- return Err(PyIOError::new_err(
- "I/O operation failed for reading on closed file.",
- ));
- }
- };
+ let mut guard = state.lock().await;
+ let writer = match guard.deref_mut() {
+ AsyncFileState::Reader(_) => {
+ return Err(PyIOError::new_err(
+ "I/O operation failed for reading on read only file.",
+ ));
+ }
+ AsyncFileState::Writer(w) => w,
+ _ => {
+ return Err(PyIOError::new_err(
+ "I/O operation failed for reading on closed file.",
+ ));
+ }
+ };
- let len = bs.len();
- writer
- .write(bs)
- .await
- .map(|_| len)
- .map_err(|err| PyIOError::new_err(err.to_string()))
- })
+ let len = bs.len();
+ writer
+ .write(bs)
+ .await
+ .map(|_| len)
+ .map_err(|err| PyIOError::new_err(err.to_string()))
}
/// Change the stream position to the given byte offset.
@@ -398,7 +393,7 @@ impl AsyncFile {
///
/// Return the new absolute position.
#[pyo3(signature = (pos, whence = 0))]
- pub fn seek<'p>(&'p mut self, py: Python<'p>, pos: i64, whence: u8) -> PyResult<&'p PyAny> {
+ pub async fn seek<'p>(&'p mut self, py: Python<'p>, pos: i64, whence: u8) -> PyResult<u64> {
let state = self.0.clone();
let whence = match whence {
@@ -408,121 +403,105 @@ impl AsyncFile {
_ => return Err(PyValueError::new_err("invalid whence")),
};
- future_into_py(py, async move {
- let mut guard = state.lock().await;
- let reader = match guard.deref_mut() {
- AsyncFileState::Reader(r) => r,
- AsyncFileState::Writer(_) => {
- return Err(PyIOError::new_err(
- "I/O operation failed for reading on write only file.",
- ));
- }
- _ => {
- return Err(PyIOError::new_err(
- "I/O operation failed for reading on closed file.",
- ));
- }
- };
+ let mut guard = state.lock().await;
+ let reader = match guard.deref_mut() {
+ AsyncFileState::Reader(r) => r,
+ AsyncFileState::Writer(_) => {
+ return Err(PyIOError::new_err(
+ "I/O operation failed for reading on write only file.",
+ ));
+ }
+ _ => {
+ return Err(PyIOError::new_err(
+ "I/O operation failed for reading on closed file.",
+ ));
+ }
+ };
- let ret = reader
- .seek(whence)
- .await
- .map_err(|err| PyIOError::new_err(err.to_string()))?;
- Ok(Python::with_gil(|py| ret.into_py(py)))
- })
+ let ret = reader
+ .seek(whence)
+ .await
+ .map_err(|err| PyIOError::new_err(err.to_string()))?;
+ Ok(ret)
}
/// Return the current stream position.
- pub fn tell<'p>(&'p mut self, py: Python<'p>) -> PyResult<&'p PyAny> {
+ pub async fn tell<'p>(&'p mut self, py: Python<'p>) -> PyResult<u64> {
let state = self.0.clone();
- future_into_py(py, async move {
- let mut guard = state.lock().await;
- let reader = match guard.deref_mut() {
- AsyncFileState::Reader(r) => r,
- AsyncFileState::Writer(_) => {
- return Err(PyIOError::new_err(
- "I/O operation failed for reading on write only file.",
- ));
- }
- _ => {
- return Err(PyIOError::new_err(
- "I/O operation failed for reading on closed file.",
- ));
- }
- };
+ let mut guard = state.lock().await;
+ let reader = match guard.deref_mut() {
+ AsyncFileState::Reader(r) => r,
+ AsyncFileState::Writer(_) => {
+ return Err(PyIOError::new_err(
+ "I/O operation failed for reading on write only file.",
+ ));
+ }
+ _ => {
+ return Err(PyIOError::new_err(
+ "I/O operation failed for reading on closed file.",
+ ));
+ }
+ };
- let pos = reader
- .stream_position()
- .await
- .map_err(|err| PyIOError::new_err(err.to_string()))?;
- Ok(Python::with_gil(|py| pos.into_py(py)))
- })
+ let pos = reader
+ .stream_position()
+ .await
+ .map_err(|err| PyIOError::new_err(err.to_string()))?;
+ Ok(pos)
}
- fn close<'p>(&'p mut self, py: Python<'p>) -> PyResult<&'p PyAny> {
+ async fn close<'p>(&'p mut self, py: Python<'p>) -> PyResult<()> {
let state = self.0.clone();
- future_into_py(py, async move {
- let mut state = state.lock().await;
- if let AsyncFileState::Writer(w) = &mut *state {
- w.close()
- .await
- .map_err(|err| PyIOError::new_err(err.to_string()))?;
- }
- *state = AsyncFileState::Closed;
- Ok(())
- })
+ let mut state = state.lock().await;
+ if let AsyncFileState::Writer(w) = &mut *state {
+ w.close()
+ .await
+ .map_err(|err| PyIOError::new_err(err.to_string()))?;
+ }
+ *state = AsyncFileState::Closed;
+ Ok(())
}
- fn __aenter__<'a>(slf: PyRef<'a, Self>, py: Python<'a>) -> PyResult<&'a PyAny> {
+ async fn __aenter__<'a>(slf: PyRef<'a, Self>, py: Python<'a>) -> PyResult<PyObject> {
let slf = slf.into_py(py);
- future_into_py(py, async move { Ok(slf) })
+ Ok(slf)
}
- fn __aexit__<'a>(
+ async fn __aexit__<'a>(
&'a mut self,
py: Python<'a>,
_exc_type: &'a PyAny,
_exc_value: &'a PyAny,
_traceback: &'a PyAny,
- ) -> PyResult<&'a PyAny> {
- self.close(py)
+ ) -> PyResult<()> {
+ self.close(py).await
}
/// Check if the stream may be read from.
- pub fn readable<'p>(&'p self, py: Python<'p>) -> PyResult<&'p PyAny> {
+ pub async fn readable<'p>(&'p self, py: Python<'p>) -> PyResult<bool> {
let state = self.0.clone();
- future_into_py(py, async move {
- let state = state.lock().await;
- Ok(matches!(*state, AsyncFileState::Reader(_)))
- })
+ let state = state.lock().await;
+ Ok(matches!(*state, AsyncFileState::Reader(_)))
}
/// Check if the stream may be written to.
- pub fn writable<'p>(&'p self, py: Python<'p>) -> PyResult<&'p PyAny> {
+ pub async fn writable<'p>(&'p self, py: Python<'p>) -> PyResult<bool> {
let state = self.0.clone();
- future_into_py(py, async move {
- let state = state.lock().await;
- Ok(matches!(*state, AsyncFileState::Writer(_)))
- })
+ let state = state.lock().await;
+ Ok(matches!(*state, AsyncFileState::Writer(_)))
}
/// Check if the stream reader may be re-located.
- pub fn seekable<'p>(&'p self, py: Python<'p>) -> PyResult<&'p PyAny> {
- if true {
- self.readable(py)
- } else {
- future_into_py(py, async move { Ok(false) })
- }
+ pub async fn seekable<'p>(&'p self, py: Python<'p>) -> PyResult<bool> {
+ self.readable(py).await
}
/// Check if the stream is closed.
#[getter]
- pub fn closed<'p>(&'p self, py: Python<'p>) -> PyResult<&'p PyAny> {
+ pub async fn closed<'p>(&'p self, py: Python<'p>) -> PyResult<bool> {
let state = self.0.clone();
- future_into_py(py, async move {
- let state = state.lock().await;
- Ok(matches!(*state, AsyncFileState::Closed))
- })
+ let state = state.lock().await;
+ Ok(matches!(*state, AsyncFileState::Closed))
}
}
diff --git a/bindings/python/src/lister.rs b/bindings/python/src/lister.rs
index e108ec2dcc..d7fdccb3c5 100644
--- a/bindings/python/src/lister.rs
+++ b/bindings/python/src/lister.rs
@@ -20,7 +20,6 @@ use std::sync::Arc;
use futures::TryStreamExt;
use pyo3::exceptions::PyStopAsyncIteration;
use pyo3::prelude::*;
-use pyo3_asyncio::tokio::future_into_py;
use tokio::sync::Mutex;
use crate::*;
@@ -66,16 +65,14 @@ impl AsyncLister {
fn __aiter__(slf: PyRef<'_, Self>) -> PyRef<Self> {
slf
}
- fn __anext__(slf: PyRefMut<'_, Self>) -> PyResult<Option<PyObject>> {
+
+ async fn __anext__(slf: PyRefMut<'_, Self>) -> PyResult<PyObject> {
let lister = slf.0.clone();
- let fut = future_into_py(slf.py(), async move {
- let mut lister = lister.lock().await;
- let entry = lister.try_next().await.map_err(format_pyerr)?;
- match entry {
- Some(entry) => Ok(Python::with_gil(|py| Entry::new(entry).into_py(py))),
- None => Err(PyStopAsyncIteration::new_err("stream exhausted")),
- }
- })?;
- Ok(Some(fut.into()))
+ let mut lister = lister.lock().await;
+ let entry = lister.try_next().await.map_err(format_pyerr)?;
+ match entry {
+ Some(entry) => Ok(Python::with_gil(|py| Entry::new(entry).into_py(py))),
+ None => Err(PyStopAsyncIteration::new_err("stream exhausted")),
+ }
}
}
diff --git a/bindings/python/src/operator.rs b/bindings/python/src/operator.rs
index 8a3425ae7a..a62e3b08d6 100644
--- a/bindings/python/src/operator.rs
+++ b/bindings/python/src/operator.rs
@@ -17,23 +17,31 @@
use std::collections::HashMap;
use std::str::FromStr;
+use std::sync::OnceLock;
use std::time::Duration;
use pyo3::exceptions::PyValueError;
use pyo3::prelude::*;
use pyo3::types::PyBytes;
use pyo3::types::PyDict;
-use pyo3_asyncio::tokio::future_into_py;
use crate::*;
+static RUNTIME: OnceLock<tokio::runtime::Runtime> = OnceLock::new();
+
fn build_operator(
scheme: ocore::Scheme,
map: HashMap<String, String>,
) -> PyResult<ocore::Operator> {
let mut op = ocore::Operator::via_map(scheme, map).map_err(format_pyerr)?;
if !op.info().full_capability().blocking {
- let runtime = pyo3_asyncio::tokio::get_runtime();
+ let runtime = RUNTIME.get_or_init(|| {
+ tokio::runtime::Builder::new_multi_thread()
+ .enable_all()
+ .build()
+ .expect("tokio runtime must be created")
+ });
+
let _guard = runtime.enter();
op = op
.layer(ocore::layers::BlockingLayer::create().expect("blocking layer must be created"));
@@ -239,111 +247,103 @@ impl AsyncOperator {
}
/// Open a file-like reader for the given path.
- pub fn open<'p>(&'p self, py: Python<'p>, path: String, mode: String) -> PyResult<&'p PyAny> {
+ pub async fn open<'p>(
+ &'p self,
+ py: Python<'p>,
+ path: String,
+ mode: String,
+ ) -> PyResult<AsyncFile> {
let this = self.0.clone();
let capability = self.capability()?;
- future_into_py(py, async move {
- if mode == "rb" {
- let meta = this.stat(&path).await.map_err(format_pyerr)?;
- let r = this.reader(&path).await.map_err(format_pyerr)?;
- Ok(AsyncFile::new_reader(r, meta.content_length(), capability))
- } else if mode == "wb" {
- let w = this.writer(&path).await.map_err(format_pyerr)?;
- Ok(AsyncFile::new_writer(w, capability))
- } else {
- Err(UnsupportedError::new_err(format!(
- "OpenDAL doesn't support mode: {mode}"
- )))
- }
- })
+ if mode == "rb" {
+ let meta = this.stat(&path).await.map_err(format_pyerr)?;
+ let r = this.reader(&path).await.map_err(format_pyerr)?;
+ Ok(AsyncFile::new_reader(r, meta.content_length(), capability))
+ } else if mode == "wb" {
+ let w = this.writer(&path).await.map_err(format_pyerr)?;
+ Ok(AsyncFile::new_writer(w, capability))
+ } else {
+ Err(UnsupportedError::new_err(format!(
+ "OpenDAL doesn't support mode: {mode}"
+ )))
+ }
}
/// Read the whole path into bytes.
- pub fn read<'p>(&'p self, py: Python<'p>, path: String) -> PyResult<&'p PyAny> {
+ pub async fn read<'p>(&'p self, py: Python<'p>, path: String) -> PyResult<Py<PyAny>> {
let this = self.0.clone();
- future_into_py(py, async move {
- let res: Vec<u8> = this.read(&path).await.map_err(format_pyerr)?.to_vec();
- Python::with_gil(|py| Buffer::new(res).into_bytes(py))
- })
+
+ let res: Vec<u8> = this.read(&path).await.map_err(format_pyerr)?.to_vec();
+ Python::with_gil(|py| Buffer::new(res).into_bytes(py))
}
/// Write bytes into given path.
#[pyo3(signature = (path, bs, **kwargs))]
- pub fn write<'p>(
+ pub async fn write<'p>(
&'p self,
py: Python<'p>,
path: String,
bs: &PyBytes,
kwargs: Option<&PyDict>,
- ) -> PyResult<&'p PyAny> {
+ ) -> PyResult<()> {
let opwrite = build_opwrite(kwargs)?;
let this = self.0.clone();
let bs = bs.as_bytes().to_vec();
- future_into_py(py, async move {
- let mut write = this.write_with(&path, bs).append(opwrite.append());
- if let Some(buffer) = opwrite.chunk() {
- write = write.chunk(buffer);
- }
- if let Some(content_type) = opwrite.content_type() {
- write = write.content_type(content_type);
- }
- if let Some(content_disposition) = opwrite.content_disposition() {
- write = write.content_disposition(content_disposition);
- }
- if let Some(cache_control) = opwrite.cache_control() {
- write = write.cache_control(cache_control);
- }
- write.await.map_err(format_pyerr)
- })
+ let mut write = this.write_with(&path, bs).append(opwrite.append());
+ if let Some(buffer) = opwrite.chunk() {
+ write = write.chunk(buffer);
+ }
+ if let Some(content_type) = opwrite.content_type() {
+ write = write.content_type(content_type);
+ }
+ if let Some(content_disposition) = opwrite.content_disposition() {
+ write = write.content_disposition(content_disposition);
+ }
+ if let Some(cache_control) = opwrite.cache_control() {
+ write = write.cache_control(cache_control);
+ }
+ write.await.map_err(format_pyerr)
}
/// Get current path's metadata **without cache** directly.
- pub fn stat<'p>(&'p self, py: Python<'p>, path: String) -> PyResult<&'p PyAny> {
+ pub async fn stat<'p>(&'p self, py: Python<'p>, path: String) -> PyResult<Metadata> {
let this = self.0.clone();
- future_into_py(py, async move {
- let res: Metadata = this
- .stat(&path)
- .await
- .map_err(format_pyerr)
- .map(Metadata::new)?;
+ let res: Metadata = this
+ .stat(&path)
+ .await
+ .map_err(format_pyerr)
+ .map(Metadata::new)?;
- Ok(res)
- })
+ Ok(res)
}
/// Copy source to target.``
- pub fn copy<'p>(
+ pub async fn copy<'p>(
&'p self,
py: Python<'p>,
source: String,
target: String,
- ) -> PyResult<&'p PyAny> {
+ ) -> PyResult<()> {
let this = self.0.clone();
- future_into_py(py, async move {
- this.copy(&source, &target).await.map_err(format_pyerr)
- })
+ this.copy(&source, &target).await.map_err(format_pyerr)
}
/// Rename filename
- pub fn rename<'p>(
+ pub async fn rename<'p>(
&'p self,
py: Python<'p>,
source: String,
target: String,
- ) -> PyResult<&'p PyAny> {
+ ) -> PyResult<()> {
let this = self.0.clone();
- future_into_py(py, async move {
- this.rename(&source, &target).await.map_err(format_pyerr)
- })
+ this.rename(&source, &target).await.map_err(format_pyerr)
}
/// Remove all file
- pub fn remove_all<'p>(&'p self, py: Python<'p>, path: String) -> PyResult<&'p PyAny> {
+ pub async fn remove_all<'p>(&'p self, py: Python<'p>, path: String) -> PyResult<()> {
let this = self.0.clone();
- future_into_py(py, async move {
- this.remove_all(&path).await.map_err(format_pyerr)
- })
+ this.remove_all(&path).await.map_err(format_pyerr)
}
/// Create a dir at given path.
@@ -358,11 +358,9 @@ impl AsyncOperator {
///
/// - Create on existing dir will succeed.
/// - Create dir is always recursive, works like `mkdir -p`
- pub fn create_dir<'p>(&'p self, py: Python<'p>, path: String) -> PyResult<&'p PyAny> {
+ pub async fn create_dir<'p>(&'p self, py: Python<'p>, path: String) -> PyResult<()> {
let this = self.0.clone();
- future_into_py(py, async move {
- this.create_dir(&path).await.map_err(format_pyerr)
- })
+ this.create_dir(&path).await.map_err(format_pyerr)
}
/// Delete given path.
@@ -370,93 +368,80 @@ impl AsyncOperator {
/// # Notes
///
/// - Delete not existing error won't return errors.
- pub fn delete<'p>(&'p self, py: Python<'p>, path: String) -> PyResult<&'p PyAny> {
+ pub async fn delete<'p>(&'p self, py: Python<'p>, path: String) -> PyResult<()> {
let this = self.0.clone();
- future_into_py(
- py,
- async move { this.delete(&path).await.map_err(format_pyerr) },
- )
+ this.delete(&path).await.map_err(format_pyerr)
}
/// List current dir path.
- pub fn list<'p>(&'p self, py: Python<'p>, path: String) -> PyResult<&'p PyAny> {
+ pub async fn list<'p>(&'p self, py: Python<'p>, path: String) -> PyResult<PyObject> {
let this = self.0.clone();
- future_into_py(py, async move {
- let lister = this.lister(&path).await.map_err(format_pyerr)?;
- let pylister: PyObject = Python::with_gil(|py| AsyncLister::new(lister).into_py(py));
- Ok(pylister)
- })
+ let lister = this.lister(&path).await.map_err(format_pyerr)?;
+ let pylister: PyObject = Python::with_gil(|py| AsyncLister::new(lister).into_py(py));
+ Ok(pylister)
}
/// List dir in flat way.
- pub fn scan<'p>(&'p self, py: Python<'p>, path: String) -> PyResult<&'p PyAny> {
+ pub async fn scan<'p>(&'p self, py: Python<'p>, path: String) -> PyResult<PyObject> {
let this = self.0.clone();
- future_into_py(py, async move {
- let lister = this
- .lister_with(&path)
- .recursive(true)
- .await
- .map_err(format_pyerr)?;
- let pylister: PyObject = Python::with_gil(|py| AsyncLister::new(lister).into_py(py));
- Ok(pylister)
- })
+ let lister = this
+ .lister_with(&path)
+ .recursive(true)
+ .await
+ .map_err(format_pyerr)?;
+ let pylister: PyObject = Python::with_gil(|py| AsyncLister::new(lister).into_py(py));
+ Ok(pylister)
}
/// Presign an operation for stat(head) which expires after `expire_second` seconds.
- pub fn presign_stat<'p>(
+ pub async fn presign_stat<'p>(
&'p self,
py: Python<'p>,
path: String,
expire_second: u64,
- ) -> PyResult<&'p PyAny> {
+ ) -> PyResult<PresignedRequest> {
let this = self.0.clone();
- future_into_py(py, async move {
- let res = this
- .presign_stat(&path, Duration::from_secs(expire_second))
- .await
- .map_err(format_pyerr)
- .map(PresignedRequest)?;
+ let res = this
+ .presign_stat(&path, Duration::from_secs(expire_second))
+ .await
+ .map_err(format_pyerr)
+ .map(PresignedRequest)?;
- Ok(res)
- })
+ Ok(res)
}
/// Presign an operation for read which expires after `expire_second` seconds.
- pub fn presign_read<'p>(
+ pub async fn presign_read<'p>(
&'p self,
py: Python<'p>,
path: String,
expire_second: u64,
- ) -> PyResult<&'p PyAny> {
+ ) -> PyResult<PresignedRequest> {
let this = self.0.clone();
- future_into_py(py, async move {
- let res = this
- .presign_read(&path, Duration::from_secs(expire_second))
- .await
- .map_err(format_pyerr)
- .map(PresignedRequest)?;
+ let res = this
+ .presign_read(&path, Duration::from_secs(expire_second))
+ .await
+ .map_err(format_pyerr)
+ .map(PresignedRequest)?;
- Ok(res)
- })
+ Ok(res)
}
/// Presign an operation for write which expires after `expire_second` seconds.
- pub fn presign_write<'p>(
+ pub async fn presign_write<'p>(
&'p self,
py: Python<'p>,
path: String,
expire_second: u64,
- ) -> PyResult<&'p PyAny> {
+ ) -> PyResult<PresignedRequest> {
let this = self.0.clone();
- future_into_py(py, async move {
- let res = this
- .presign_write(&path, Duration::from_secs(expire_second))
- .await
- .map_err(format_pyerr)
- .map(PresignedRequest)?;
-
- Ok(res)
- })
+ let res = this
+ .presign_write(&path, Duration::from_secs(expire_second))
+ .await
+ .map_err(format_pyerr)
+ .map(PresignedRequest)?;
+
+ Ok(res)
}
pub fn capability(&self) -> PyResult<capability::Capability> {