feat(bindings/python): expose DeleteOptions version and recursive params

bindings/python/src/capability.rs

@@ -96,6 +96,10 @@ pub struct Capability {
 
     /// If operator supports delete.
     pub delete: bool,
+    /// If delete operations with version are supported.
+    pub delete_with_version: bool,
+    /// If recursive delete operations are supported.
+    pub delete_with_recursive: bool,
 
     /// If operator supports copy.
     pub copy: bool,
@@ -160,6 +164,8 @@ impl Capability {
             write_with_user_metadata: capability.write_with_user_metadata,
             create_dir: capability.create_dir,
             delete: capability.delete,
+            delete_with_version: capability.delete_with_version,
+            delete_with_recursive: capability.delete_with_recursive,
             copy: capability.copy,
             rename: capability.rename,
             list: capability.list,

bindings/python/src/operator.rs

@@ -523,9 +523,27 @@ impl Operator {
     /// ----------
     /// path : str
     ///     The path to the file.
-    pub fn delete(&self, path: PathBuf) -> PyResult<()> {
+    /// version : str, optional
+    ///     The version of the file to delete. Only supported on version-aware backends.
+    /// recursive : bool, optional
+    ///     If True, delete the path recursively. Only supported on backends that support recursive delete.
+    #[pyo3(signature = (path, *, version=None, recursive=None))]
+    pub fn delete(
+        &self,
+        path: PathBuf,
+        version: Option<String>,
+        recursive: Option<bool>,
+    ) -> PyResult<()> {
         let path = path.to_string_lossy().to_string();
-        self.core.delete(&path).map_err(format_pyerr)
+        if version.is_some() || recursive.is_some() {
+            let opts = ocore::options::DeleteOptions {
+                version,
+                recursive: recursive.unwrap_or(false),
+            };
+            self.core.delete_options(&path, opts).map_err(format_pyerr)
+        } else {
+            self.core.delete(&path).map_err(format_pyerr)
+        }
     }
 
     /// Check if a path exists.
@@ -1302,13 +1320,31 @@ impl AsyncOperator {
         type_repr="collections.abc.Awaitable[None]",
         imports=("collections.abc")
     ))]
-    pub fn delete<'p>(&'p self, py: Python<'p>, path: PathBuf) -> PyResult<Bound<'p, PyAny>> {
+    /// version : str, optional
+    ///     The version of the file to delete. Only supported on version-aware backends.
+    /// recursive : bool, optional
+    ///     If True, delete the path recursively. Only supported on backends that support recursive delete.
+    #[pyo3(signature = (path, *, version=None, recursive=None))]
+    pub fn delete<'p>(
+        &'p self,
+        py: Python<'p>,
+        path: PathBuf,
+        version: Option<String>,
+        recursive: Option<bool>,
+    ) -> PyResult<Bound<'p, PyAny>> {
         let this = self.core.clone();
         let path = path.to_string_lossy().to_string();
-        future_into_py(
-            py,
-            async move { this.delete(&path).await.map_err(format_pyerr) },
-        )
+        future_into_py(py, async move {
+            if version.is_some() || recursive.is_some() {
+                let opts = ocore::options::DeleteOptions {
+                    version,
+                    recursive: recursive.unwrap_or(false),
+                };
+                this.delete_options(&path, opts).await.map_err(format_pyerr)
+            } else {
+                this.delete(&path).await.map_err(format_pyerr)
+            }
+        })
     }
 
     /// Check if a path exists.
@@ -1718,7 +1754,10 @@ impl AsyncOperator {
     ) -> PyResult<Bound<'p, PyAny>> {
         let this = self.core.clone();
         let path = path.to_string_lossy().to_string();
-        let opts = DeleteOptions { version };
+        let opts = DeleteOptions {
+            version,
+            ..Default::default()
+        };
         future_into_py(py, async move {
             let res = this
                 .presign_delete_options(&path, Duration::from_secs(expire_second), opts.into())

bindings/python/src/options.rs

@@ -286,13 +286,27 @@ impl From<StatOptions> for ocore::options::StatOptions {
 #[derive(Default, Debug)]
 pub struct DeleteOptions {
     pub version: Option<String>,
+    pub recursive: Option<bool>,
+}
+
+impl<'a, 'py> FromPyObject<'a, 'py> for DeleteOptions {
+    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")?,
+            recursive: extract_optional(&dict, "recursive")?,
+        })
+    }
 }
 
 impl From<DeleteOptions> for ocore::options::DeleteOptions {
     fn from(opts: DeleteOptions) -> Self {
         Self {
             version: opts.version,
-            ..Default::default()
+            recursive: opts.recursive.unwrap_or(false),
         }
     }
 }

bindings/python/tests/test_delete_options.py

@@ -0,0 +1,61 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+from uuid import uuid4
+
+import pytest
+
+
+@pytest.mark.need_capability("write", "delete", "delete_with_version")
+def test_delete_accepts_version_param(service_name, operator, async_operator):
+    path = f"test_delete_version_{uuid4()}.txt"
+    operator.write(path, b"test content")
+    operator.delete(path, version="v1")
+
+
+@pytest.mark.need_capability("write", "delete", "delete_with_recursive")
+def test_delete_accepts_recursive_param(service_name, operator, async_operator):
+    path = f"test_delete_recursive_{uuid4()}.txt"
+    operator.write(path, b"test content")
+    operator.delete(path, recursive=True)
+
+
+@pytest.mark.need_capability("write", "delete")
+def test_delete_default_params_unchanged(service_name, operator, async_operator):
+    path = f"test_delete_default_{uuid4()}.txt"
+    operator.write(path, b"test content")
+    operator.delete(path)
+
+
+@pytest.mark.need_capability("write", "delete", "delete_with_version")
+@pytest.mark.asyncio
+async def test_async_delete_accepts_version_param(
+    service_name, operator, async_operator
+):
+    path = f"test_async_delete_version_{uuid4()}.txt"
+    await async_operator.write(path, b"test content")
+    await async_operator.delete(path, version="v1")
+
+
+@pytest.mark.need_capability("write", "delete", "delete_with_recursive")
+@pytest.mark.asyncio
+async def test_async_delete_accepts_recursive_param(
+    service_name, operator, async_operator
+):
+    path = f"test_async_delete_recursive_{uuid4()}.txt"
+    await async_operator.write(path, b"test content")
+    await async_operator.delete(path, recursive=True)