feat(bindings/python): export ConcurrentLimitLayer (#5140)

Author: TennyZhuang

bindings/python/README.md

@@ -41,7 +41,7 @@ async def main():
 asyncio.run(main())
 ```
 
-s3 service example: 
+s3 service example:
 ```python
 import opendal
 

bindings/python/python/opendal/layers.pyi

@@ -30,3 +30,7 @@ class RetryLayer(Layer):
         max_delay: Optional[float] = None,
         min_delay: Optional[float] = None,
     ) -> None: ...
+
+@final
+class ConcurrentLimitLayer(Layer):
+    def __init__(self, limit: int) -> None: ...

bindings/python/src/layers.rs

@@ -80,3 +80,26 @@ impl RetryLayer {
         Ok(class)
     }
 }
+
+#[pyclass(module = "opendal.layers", extends=Layer)]
+#[derive(Clone)]
+pub struct ConcurrentLimitLayer(ocore::layers::ConcurrentLimitLayer);
+
+impl PythonLayer for ConcurrentLimitLayer {
+    fn layer(&self, op: Operator) -> Operator {
+        op.layer(self.0.clone())
+    }
+}
+
+#[pymethods]
+impl ConcurrentLimitLayer {
+    #[new]
+    #[pyo3(signature = (limit))]
+    fn new(limit: usize) -> PyResult<PyClassInitializer<Self>> {
+        let concurrent_limit = Self(ocore::layers::ConcurrentLimitLayer::new(limit));
+        let class = PyClassInitializer::from(Layer(Box::new(concurrent_limit.clone())))
+            .add_subclass(concurrent_limit);
+
+        Ok(class)
+    }
+}

bindings/python/src/lib.rs

@@ -90,6 +90,7 @@ fn _opendal(py: Python, m: &Bound<'_, PyModule>) -> PyResult<()> {
     let layers_module = PyModule::new_bound(py, "layers")?;
     layers_module.add_class::<Layer>()?;
     layers_module.add_class::<RetryLayer>()?;
+    layers_module.add_class::<ConcurrentLimitLayer>()?;
     m.add_submodule(&layers_module)?;
     py.import_bound("sys")?
         .getattr("modules")?

bindings/python/tests/conftest.py

@@ -60,8 +60,10 @@ def setup_config(service_name):
 
 @pytest.fixture(scope="session")
 def async_operator(service_name, setup_config):
-    return opendal.AsyncOperator(service_name, **setup_config).layer(
-        opendal.layers.RetryLayer()
+    return (
+        opendal.AsyncOperator(service_name, **setup_config)
+        .layer(opendal.layers.RetryLayer())
+        .layer(opendal.layers.ConcurrentLimitLayer(1024))
     )