Merge branch 'main' into feature/s3

Author: sksingh2005

bindings/python/Cargo.toml

@@ -198,7 +198,6 @@ name = "_opendal"
 
 [dependencies]
 bytes = "1.5.0"
-dict_derive = "0.6.0"
 futures = "0.3.28"
 jiff = { version = "0.2.15" }
 mea = "0.6"
@@ -207,8 +206,8 @@ opendal = { version = ">=0", path = "../../core", features = [
   "blocking",
   "layers-mime-guess",
 ] }
-pyo3 = { version = "0.26.0", features = ["generate-import-lib", "jiff-02"] }
-pyo3-async-runtimes = { version = "0.26.0", features = ["tokio-runtime"] }
+pyo3 = { version = "0.27.2", features = ["generate-import-lib", "jiff-02"] }
+pyo3-async-runtimes = { version = "0.27.0", features = ["tokio-runtime"] }
 pyo3-stub-gen = { version = "0.17" }
 tokio = "1"
 

bindings/python/src/options.rs

@@ -15,13 +15,22 @@
 // specific language governing permissions and limitations
 // under the License.
 
-use dict_derive::FromPyObject;
 use opendal::{self as ocore, raw::BytesRange};
+use pyo3::Borrowed;
+use pyo3::FromPyObject;
+use pyo3::PyAny;
+use pyo3::PyErr;
+use pyo3::PyResult;
+use pyo3::conversion::FromPyObjectOwned;
+use pyo3::exceptions::PyTypeError;
 use pyo3::pyclass;
+use pyo3::types::PyAnyMethods;
+use pyo3::types::PyDict;
+use pyo3::types::PyDictMethods;
 use std::collections::HashMap;
 
 #[pyclass(module = "opendal")]
-#[derive(FromPyObject, Default)]
+#[derive(Default)]
 pub struct ReadOptions {
     pub version: Option<String>,
     pub concurrent: Option<usize>,
@@ -39,6 +48,55 @@ pub struct ReadOptions {
     pub content_disposition: Option<String>,
 }
 
+fn map_exception(name: &str, err: PyErr) -> PyErr {
+    PyErr::new::<PyTypeError, _>(format!("Unable to convert key: {name}. Error: {err}"))
+}
+
+fn extract_optional<'py, T>(dict: &pyo3::Bound<'py, PyDict>, name: &str) -> PyResult<Option<T>>
+where
+    T: FromPyObjectOwned<'py>,
+{
+    match dict.get_item(name)? {
+        Some(v) => v
+            .extract::<T>()
+            .map(Some)
+            .map_err(|err| map_exception(name, err.into())),
+        None => Ok(None),
+    }
+}
+
+fn downcast_kwargs<'a, 'py>(obj: Borrowed<'a, 'py, PyAny>) -> PyResult<pyo3::Bound<'py, PyDict>> {
+    let obj: &pyo3::Bound<'_, PyAny> = &obj;
+    obj.cast::<PyDict>()
+        .cloned()
+        .map_err(|_| PyErr::new::<PyTypeError, _>("Invalid type to convert, expected dict"))
+}
+
+impl<'a, 'py> FromPyObject<'a, 'py> for ReadOptions {
+    type Error = PyErr;
+
+    fn extract(obj: Borrowed<'a, 'py, PyAny>) -> Result<Self, Self::Error> {
+        let dict = downcast_kwargs(obj)?;
+
+        Ok(Self {
+            version: extract_optional(&dict, "version")?,
+            concurrent: extract_optional(&dict, "concurrent")?,
+            chunk: extract_optional(&dict, "chunk")?,
+            gap: extract_optional(&dict, "gap")?,
+            offset: extract_optional(&dict, "offset")?,
+            prefetch: extract_optional(&dict, "prefetch")?,
+            size: extract_optional(&dict, "size")?,
+            if_match: extract_optional(&dict, "if_match")?,
+            if_none_match: extract_optional(&dict, "if_none_match")?,
+            if_modified_since: extract_optional(&dict, "if_modified_since")?,
+            if_unmodified_since: extract_optional(&dict, "if_unmodified_since")?,
+            content_type: extract_optional(&dict, "content_type")?,
+            cache_control: extract_optional(&dict, "cache_control")?,
+            content_disposition: extract_optional(&dict, "content_disposition")?,
+        })
+    }
+}
+
 impl ReadOptions {
     pub fn make_range(&self) -> BytesRange {
         let offset = self.offset.unwrap_or_default() as u64;
@@ -49,7 +107,7 @@ impl ReadOptions {
 }
 
 #[pyclass(module = "opendal")]
-#[derive(FromPyObject, Default)]
+#[derive(Default)]
 pub struct WriteOptions {
     pub append: Option<bool>,
     pub chunk: Option<usize>,
@@ -64,6 +122,28 @@ pub struct WriteOptions {
     pub user_metadata: Option<HashMap<String, String>>,
 }
 
+impl<'a, 'py> FromPyObject<'a, 'py> for WriteOptions {
+    type Error = PyErr;
+
+    fn extract(obj: Borrowed<'a, 'py, PyAny>) -> Result<Self, Self::Error> {
+        let dict = downcast_kwargs(obj)?;
+
+        Ok(Self {
+            append: extract_optional(&dict, "append")?,
+            chunk: extract_optional(&dict, "chunk")?,
+            concurrent: extract_optional(&dict, "concurrent")?,
+            cache_control: extract_optional(&dict, "cache_control")?,
+            content_type: extract_optional(&dict, "content_type")?,
+            content_disposition: extract_optional(&dict, "content_disposition")?,
+            content_encoding: extract_optional(&dict, "content_encoding")?,
+            if_match: extract_optional(&dict, "if_match")?,
+            if_none_match: extract_optional(&dict, "if_none_match")?,
+            if_not_exists: extract_optional(&dict, "if_not_exists")?,
+            user_metadata: extract_optional(&dict, "user_metadata")?,
+        })
+    }
+}
+
 impl From<ReadOptions> for ocore::options::ReadOptions {
     fn from(opts: ReadOptions) -> Self {
         Self {
@@ -118,7 +198,7 @@ impl From<WriteOptions> for ocore::options::WriteOptions {
 }
 
 #[pyclass(module = "opendal")]
-#[derive(FromPyObject, Default, Debug)]
+#[derive(Default, Debug)]
 pub struct ListOptions {
     pub limit: Option<usize>,
     pub start_after: Option<String>,
@@ -127,6 +207,22 @@ pub struct ListOptions {
     pub deleted: Option<bool>,
 }
 
+impl<'a, 'py> FromPyObject<'a, 'py> for ListOptions {
+    type Error = PyErr;
+
+    fn extract(obj: Borrowed<'a, 'py, PyAny>) -> Result<Self, Self::Error> {
+        let dict = downcast_kwargs(obj)?;
+
+        Ok(Self {
+            limit: extract_optional(&dict, "limit")?,
+            start_after: extract_optional(&dict, "start_after")?,
+            recursive: extract_optional(&dict, "recursive")?,
+            versions: extract_optional(&dict, "versions")?,
+            deleted: extract_optional(&dict, "deleted")?,
+        })
+    }
+}
+
 impl From<ListOptions> for ocore::options::ListOptions {
     fn from(opts: ListOptions) -> Self {
         Self {
@@ -140,7 +236,7 @@ impl From<ListOptions> for ocore::options::ListOptions {
 }
 
 #[pyclass(module = "opendal")]
-#[derive(FromPyObject, Default, Debug)]
+#[derive(Default, Debug)]
 pub struct StatOptions {
     pub version: Option<String>,
     pub if_match: Option<String>,
@@ -152,6 +248,25 @@ pub struct StatOptions {
     pub content_disposition: Option<String>,
 }
 
+impl<'a, 'py> FromPyObject<'a, 'py> for StatOptions {
+    type Error = PyErr;
+
+    fn extract(obj: Borrowed<'a, 'py, PyAny>) -> Result<Self, Self::Error> {
+        let dict = downcast_kwargs(obj)?;
+
+        Ok(Self {
+            version: extract_optional(&dict, "version")?,
+            if_match: extract_optional(&dict, "if_match")?,
+            if_none_match: extract_optional(&dict, "if_none_match")?,
+            if_modified_since: extract_optional(&dict, "if_modified_since")?,
+            if_unmodified_since: extract_optional(&dict, "if_unmodified_since")?,
+            content_type: extract_optional(&dict, "content_type")?,
+            cache_control: extract_optional(&dict, "cache_control")?,
+            content_disposition: extract_optional(&dict, "content_disposition")?,
+        })
+    }
+}
+
 impl From<StatOptions> for ocore::options::StatOptions {
     fn from(opts: StatOptions) -> Self {
         Self {

core/core/src/layers/complete.rs

@@ -153,10 +153,32 @@ impl<R: oio::Read> oio::Read for CompleteReader<R> {
     }
 }
 
+/// Tracks the state of the Write operation.
+/// A successful operation goes through states: Open -> Written -> Closed
+/// A failed operation terminates in the Error state
+#[derive(Debug, PartialEq, Eq)]
+enum CompleteState {
+    Open,
+    Written,
+    Closed,
+    Error,
+}
+
+impl CompleteState {
+    /// Attempt to transition to the destination state. Once CompleteState has
+    /// errored all further transitions are ignored.
+    fn transition(&mut self, destination: CompleteState) {
+        if *self != CompleteState::Error {
+            *self = destination
+        }
+    }
+}
+
 pub struct CompleteWriter<W> {
     inner: Option<W>,
     append: bool,
     size: u64,
+    state: CompleteState,
 }
 
 impl<W> CompleteWriter<W> {
@@ -165,6 +187,7 @@ impl<W> CompleteWriter<W> {
             inner: Some(inner),
             append,
             size: 0,
+            state: CompleteState::Open,
         }
     }
 
@@ -194,8 +217,10 @@ impl<W> CompleteWriter<W> {
 #[cfg(debug_assertions)]
 impl<W> Drop for CompleteWriter<W> {
     fn drop(&mut self) {
-        if self.inner.is_some() {
-            log::warn!("writer has not been closed or aborted, must be a bug")
+        if self.state == CompleteState::Written {
+            log::warn!(
+                "writer has not been closed or aborted after successful write operation, must be a bug"
+            )
         }
     }
 }
@@ -206,29 +231,47 @@ where
 {
     async fn write(&mut self, bs: Buffer) -> Result<()> {
         let w = self.inner.as_mut().ok_or_else(|| {
+            debug_assert_ne!(
+                self.state,
+                CompleteState::Open,
+                "bug: inner is empty, but state is Open"
+            );
             Error::new(ErrorKind::Unexpected, "writer has been closed or aborted")
         })?;
 
         let len = bs.len();
-        w.write(bs).await?;
+        w.write(bs)
+            .await
+            .inspect_err(|_| self.state.transition(CompleteState::Error))?;
         self.size += len as u64;
+        self.state.transition(CompleteState::Written);
 
         Ok(())
     }
 
     async fn close(&mut self) -> Result<Metadata> {
         let w = self.inner.as_mut().ok_or_else(|| {
+            debug_assert_ne!(
+                self.state,
+                CompleteState::Open,
+                "bug: inner is empty, but state is Open"
+            );
             Error::new(ErrorKind::Unexpected, "writer has been closed or aborted")
         })?;
 
         // we must return `Err` before setting inner to None; otherwise,
         // we won't be able to retry `close` in `RetryLayer`.
-        let mut ret = w.close().await?;
-        self.check(ret.content_length())?;
+        let mut ret = w
+            .close()
+            .await
+            .inspect_err(|_| self.state.transition(CompleteState::Error))?;
+        self.check(ret.content_length())
+            .inspect_err(|_| self.state.transition(CompleteState::Error))?;
         if ret.content_length() == 0 {
             ret = ret.with_content_length(self.size);
         }
         self.inner = None;
+        self.state.transition(CompleteState::Closed);
 
         Ok(ret)
     }
@@ -238,8 +281,11 @@ where
             Error::new(ErrorKind::Unexpected, "writer has been closed or aborted")
         })?;
 
-        w.abort().await?;
+        w.abort()
+            .await
+            .inspect_err(|_| self.state.transition(CompleteState::Error))?;
         self.inner = None;
+        self.state.transition(CompleteState::Closed);
 
         Ok(())
     }