This is an automated email from the ASF dual-hosted git repository.
chitralverma pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/opendal.git
The following commit(s) were added to refs/heads/main by this push:
new 54b8392b6 feat(bindings/python): add from_uri constructor for Operator
and AsyncOperator (#7869)
54b8392b6 is described below
commit 54b8392b6e2a26a74dc037e8878729aae4d2eb40
Author: Chitral Verma <[email protected]>
AuthorDate: Tue Jul 7 13:48:45 2026 +0530
feat(bindings/python): add from_uri constructor for Operator and
AsyncOperator (#7869)
* feat(bindings/python): add from_uri constructor for Operator and
AsyncOperator
Expose Operator.from_uri / AsyncOperator.from_uri classmethods backed by
the core Operator::from_uri URI registry path, mirroring the Node.js
binding. Replace __getnewargs_ex__ with __reduce__ so URI-built operators
pickle via a dedicated reconstructor instead of the scheme-based __new__.
* fix(bindings/python): address review feedback on from_uri
- Extract kwargs via extract_kwargs helper that returns a PyErr instead
of panicking on non-str values; apply to from_uri and __new__ for both
operator classes.
- Mirror the async_operator fixture layers and capability overrides in the
from_uri test fixture so capability gating and equality assertions match.
* refactor(bindings/python): simplify from_uri pickling and docs per review
- Drop the __from_uri field: reconstruct both scheme- and URI-built
operators through a single from_uri-based reconstructor, since the core
resolves bare schemes and full URIs through the same path.
- Reframe from_uri docstrings: options belong in the URI query string
(urlencode example); **kwargs documented as override-only.
* docs(bindings/python): debloat from_uri docstrings and reconstructor
comments
* docs(bindings/python): use fenced code block in from_uri examples
Replace >>> doctest prompts with a python code block so copying from the
rendered mkdocs docs does not include the prompt characters.
* refactor(bindings/python): extract kwargs directly and add from_uri
round-trip test
- Bind **kwargs to Option<HashMap<String, String>> in the operator
constructors, dropping the extract_kwargs helper; PyO3 raises TypeError
natively on non-str values.
- Add sync and async from_uri write/read round-trip tests.
---
bindings/python/python/opendal/_opendal.pyi | 18 +++
bindings/python/python/opendal/operator.pyi | 74 +++++++++++-
bindings/python/src/lib.rs | 6 +
bindings/python/src/operator.rs | 179 ++++++++++++++++++++++++----
bindings/python/tests/test_from_uri.py | 138 +++++++++++++++++++++
5 files changed, 389 insertions(+), 26 deletions(-)
diff --git a/bindings/python/python/opendal/_opendal.pyi
b/bindings/python/python/opendal/_opendal.pyi
index d5c507f3e..76695653e 100644
--- a/bindings/python/python/opendal/_opendal.pyi
+++ b/bindings/python/python/opendal/_opendal.pyi
@@ -19,6 +19,8 @@ from typing import Final, final
from _typeshed import Incomplete
+from .operator import AsyncOperator, Operator
+
__version__: Final[str]
@final
@@ -36,4 +38,20 @@ class StatOptions: ...
@final
class WriteOptions: ...
+def _reconstruct_async_operator(scheme: str, map: dict[str, str]) ->
AsyncOperator:
+ """
+ Rebuild an [`AsyncOperator`] while unpickling.
+
+ See [`_reconstruct_operator`] for why a dedicated reconstructor is used.
+ """
+
+def _reconstruct_operator(scheme: str, map: dict[str, str]) -> Operator:
+ """
+ Rebuild a blocking [`Operator`] while unpickling.
+
+ Routes through `from_uri`, not the scheme-based `__new__`, whose scheme
+ normalization would corrupt a URI held in `__scheme`. Bare schemes work
too:
+ the core resolves both through the same path.
+ """
+
def __getattr__(name: str) -> Incomplete: ...
diff --git a/bindings/python/python/opendal/operator.pyi
b/bindings/python/python/opendal/operator.pyi
index af2d60566..40c7731df 100644
--- a/bindings/python/python/opendal/operator.pyi
+++ b/bindings/python/python/opendal/operator.pyi
@@ -38,7 +38,6 @@ class AsyncOperator:
Operator
"""
- def __getnewargs_ex__(self, /) -> Any: ...
def __new__(cls, /, scheme: str | Scheme, **kwargs) -> AsyncOperator:
"""
Create a new `AsyncOperator`.
@@ -55,6 +54,7 @@ class AsyncOperator:
AsyncOperator
The new async operator.
"""
+ def __reduce__(self, /) -> Any: ...
def capability(self, /) -> Capability:
"""
Get all capabilities of this operator.
@@ -161,6 +161,42 @@ class AsyncOperator:
coroutine
An awaitable that returns True if the path exists, False otherwise.
"""
+ @classmethod
+ def from_uri(cls, /, uri: str, **kwargs) -> AsyncOperator:
+ """
+ Create a new `AsyncOperator` from a URI string.
+
+ The URI encodes the scheme and configuration in a single string, e.g.
+ ``memory://`` or ``s3://bucket/path?region=us-east-1``. The scheme must
+ belong to a service enabled in this build. Encode service options as
+ query parameters; use ``urllib.parse.urlencode`` when building the URI
+ dynamically.
+
+ Parameters
+ ----------
+ uri : str
+ The URI of the service, including any options as query parameters.
+ **kwargs : dict
+ Overrides for URI options. Prefer the URI query string.
+
+ Returns
+ -------
+ AsyncOperator
+ The new async operator.
+
+ Examples
+ --------
+ ```python
+ from urllib.parse import urlencode
+ import opendal
+
+ op = opendal.AsyncOperator.from_uri("memory://")
+ query = urlencode({"region": "us-east-1"})
+ op = opendal.AsyncOperator.from_uri(
+ f"s3://bucket/path?{query}"
+ )
+ ```
+ """
def layer(self, /, layer: Layer) -> AsyncOperator:
"""
Add a new layer to the operator.
@@ -647,7 +683,6 @@ class Operator:
AsyncOperator
"""
- def __getnewargs_ex__(self, /) -> Any: ...
def __new__(cls, /, scheme: str | Scheme, **kwargs) -> Operator:
"""
Create a new blocking `Operator`.
@@ -664,6 +699,7 @@ class Operator:
Operator
The new operator.
"""
+ def __reduce__(self, /) -> Any: ...
def capability(self, /) -> Capability:
"""
Get all capabilities of this operator.
@@ -746,6 +782,40 @@ class Operator:
bool
True if the path exists, False otherwise.
"""
+ @classmethod
+ def from_uri(cls, /, uri: str, **kwargs) -> Operator:
+ """
+ Create a new blocking `Operator` from a URI string.
+
+ The URI encodes the scheme and configuration in a single string, e.g.
+ ``memory://`` or ``s3://bucket/path?region=us-east-1``. The scheme must
+ belong to a service enabled in this build. Encode service options as
+ query parameters; use ``urllib.parse.urlencode`` when building the URI
+ dynamically.
+
+ Parameters
+ ----------
+ uri : str
+ The URI of the service, including any options as query parameters.
+ **kwargs : dict
+ Overrides for URI options. Prefer the URI query string.
+
+ Returns
+ -------
+ Operator
+ The new operator.
+
+ Examples
+ --------
+ ```python
+ from urllib.parse import urlencode
+ import opendal
+
+ op = opendal.Operator.from_uri("memory://")
+ query = urlencode({"region": "us-east-1"})
+ op = opendal.Operator.from_uri(f"s3://bucket/path?{query}")
+ ```
+ """
def layer(self, /, layer: Layer) -> Operator:
"""
Add a new layer to this operator.
diff --git a/bindings/python/src/lib.rs b/bindings/python/src/lib.rs
index f7fa61753..b923c76b0 100644
--- a/bindings/python/src/lib.rs
+++ b/bindings/python/src/lib.rs
@@ -57,6 +57,12 @@ mod _opendal {
use crate::{AsyncOperator, Operator};
}
+ // Pickle reconstructors for operators built via `from_uri`. They must live
+ // on the top-level `_opendal` module so `__reduce__` can reference them by
+ // their qualified name during unpickling.
+ #[pymodule_export]
+ use crate::{_reconstruct_async_operator, _reconstruct_operator};
+
#[pymodule]
mod file {
#[pymodule_export]
diff --git a/bindings/python/src/operator.rs b/bindings/python/src/operator.rs
index b67723589..9782e0385 100644
--- a/bindings/python/src/operator.rs
+++ b/bindings/python/src/operator.rs
@@ -24,6 +24,7 @@ use pyo3::prelude::*;
use pyo3::types::PyBytes;
use pyo3::types::PyDict;
use pyo3::types::PyTuple;
+use pyo3::types::PyType;
use pyo3_async_runtimes::tokio::future_into_py;
use crate::*;
@@ -45,10 +46,56 @@ fn build_blocking_operator(
Ok(op)
}
+fn build_operator_from_uri(uri: &str, map: HashMap<String, String>) ->
PyResult<ocore::Operator> {
+ let op = ocore::Operator::from_uri((uri, map)).map_err(format_pyerr)?;
+ Ok(op)
+}
+
+fn build_blocking_operator_from_uri(
+ uri: &str,
+ map: HashMap<String, String>,
+) -> PyResult<ocore::blocking::Operator> {
+ let op = build_operator_from_uri(uri, map)?;
+
+ let runtime = pyo3_async_runtimes::tokio::get_runtime();
+ let _guard = runtime.enter();
+ let op = ocore::blocking::Operator::new(op).map_err(format_pyerr)?;
+ Ok(op)
+}
+
fn normalize_scheme(raw: &str) -> String {
raw.trim().to_ascii_lowercase().replace('_', "-")
}
+/// Rebuild a blocking [`Operator`] while unpickling.
+///
+/// Routes through `from_uri`, not the scheme-based `__new__`, whose scheme
+/// normalization would corrupt a URI held in `__scheme`. Bare schemes work
too:
+/// the core resolves both through the same path.
+#[pyfunction]
+pub fn _reconstruct_operator(scheme: &str, map: HashMap<String, String>) ->
PyResult<Operator> {
+ Ok(Operator {
+ core: build_blocking_operator_from_uri(scheme, map.clone())?,
+ __scheme: scheme.to_string(),
+ __map: map,
+ })
+}
+
+/// Rebuild an [`AsyncOperator`] while unpickling.
+///
+/// See [`_reconstruct_operator`] for why a dedicated reconstructor is used.
+#[pyfunction]
+pub fn _reconstruct_async_operator(
+ scheme: &str,
+ map: HashMap<String, String>,
+) -> PyResult<AsyncOperator> {
+ Ok(AsyncOperator {
+ core: build_operator_from_uri(scheme, map.clone())?,
+ __scheme: scheme.to_string(),
+ __map: map,
+ })
+}
+
/// The blocking equivalent of `AsyncOperator`.
///
/// `Operator` is the entry point for all blocking APIs.
@@ -79,7 +126,7 @@ impl Operator {
/// The new operator.
#[new]
#[pyo3(signature = (scheme: "str | Scheme", *, **kwargs))]
- pub fn new(scheme: Bound<PyAny>, kwargs: Option<&Bound<PyDict>>) ->
PyResult<Self> {
+ pub fn new(scheme: Bound<PyAny>, kwargs: Option<HashMap<String, String>>)
-> PyResult<Self> {
let scheme = if let Ok(scheme_str) = scheme.extract::<&str>() {
scheme_str.to_string()
} else if let Ok(py_scheme) = scheme.extract::<Scheme>() {
@@ -90,12 +137,7 @@ impl Operator {
));
};
let scheme = normalize_scheme(&scheme);
- let map = kwargs
- .map(|v| {
- v.extract::<HashMap<String, String>>()
- .expect("must be valid hashmap")
- })
- .unwrap_or_default();
+ let map = kwargs.unwrap_or_default();
Ok(Operator {
core: build_blocking_operator(&scheme, map.clone())?,
@@ -104,6 +146,52 @@ impl Operator {
})
}
+ /// Create a new blocking `Operator` from a URI string.
+ ///
+ /// The URI encodes the scheme and configuration in a single string, e.g.
+ /// ``memory://`` or ``s3://bucket/path?region=us-east-1``. The scheme must
+ /// belong to a service enabled in this build. Encode service options as
+ /// query parameters; use ``urllib.parse.urlencode`` when building the URI
+ /// dynamically.
+ ///
+ /// Parameters
+ /// ----------
+ /// uri : str
+ /// The URI of the service, including any options as query parameters.
+ /// **kwargs : dict
+ /// Overrides for URI options. Prefer the URI query string.
+ ///
+ /// Returns
+ /// -------
+ /// Operator
+ /// The new operator.
+ ///
+ /// Examples
+ /// --------
+ /// ```python
+ /// from urllib.parse import urlencode
+ /// import opendal
+ ///
+ /// op = opendal.Operator.from_uri("memory://")
+ /// query = urlencode({"region": "us-east-1"})
+ /// op = opendal.Operator.from_uri(f"s3://bucket/path?{query}")
+ /// ```
+ #[classmethod]
+ #[pyo3(signature = (uri, **kwargs))]
+ pub fn from_uri(
+ _cls: &Bound<PyType>,
+ uri: &str,
+ kwargs: Option<HashMap<String, String>>,
+ ) -> PyResult<Self> {
+ let map = kwargs.unwrap_or_default();
+
+ Ok(Operator {
+ core: build_blocking_operator_from_uri(uri, map.clone())?,
+ __scheme: uri.to_string(),
+ __map: map,
+ })
+ }
+
/// Add a new layer to this operator.
///
/// Parameters
@@ -692,11 +780,12 @@ impl Operator {
)
}
}
- fn __getnewargs_ex__(&self, py: Python) -> PyResult<Py<PyAny>> {
- let args = vec![self.__scheme.clone()];
- let args = PyTuple::new(py, args)?.into_py_any(py)?;
- let kwargs = self.__map.clone().into_py_any(py)?;
- PyTuple::new(py, [args, kwargs])?.into_py_any(py)
+ fn __reduce__(&self, py: Python) -> PyResult<Py<PyAny>> {
+ let reconstructor = py
+ .import("opendal._opendal")?
+ .getattr("_reconstruct_operator")?;
+ let args = (self.__scheme.clone(),
self.__map.clone()).into_py_any(py)?;
+ PyTuple::new(py, [reconstructor.into_py_any(py)?,
args])?.into_py_any(py)
}
}
@@ -730,7 +819,7 @@ impl AsyncOperator {
/// The new async operator.
#[new]
#[pyo3(signature = (scheme: "str | Scheme", * ,**kwargs))]
- pub fn new(scheme: Bound<PyAny>, kwargs: Option<&Bound<PyDict>>) ->
PyResult<Self> {
+ pub fn new(scheme: Bound<PyAny>, kwargs: Option<HashMap<String, String>>)
-> PyResult<Self> {
let scheme = if let Ok(scheme_str) = scheme.extract::<&str>() {
scheme_str.to_string()
} else if let Ok(py_scheme) = scheme.extract::<Scheme>() {
@@ -742,12 +831,7 @@ impl AsyncOperator {
};
let scheme = normalize_scheme(&scheme);
- let map = kwargs
- .map(|v| {
- v.extract::<HashMap<String, String>>()
- .expect("must be valid hashmap")
- })
- .unwrap_or_default();
+ let map = kwargs.unwrap_or_default();
Ok(AsyncOperator {
core: build_operator(&scheme, map.clone())?,
@@ -756,6 +840,52 @@ impl AsyncOperator {
})
}
+ /// Create a new `AsyncOperator` from a URI string.
+ ///
+ /// The URI encodes the scheme and configuration in a single string, e.g.
+ /// ``memory://`` or ``s3://bucket/path?region=us-east-1``. The scheme must
+ /// belong to a service enabled in this build. Encode service options as
+ /// query parameters; use ``urllib.parse.urlencode`` when building the URI
+ /// dynamically.
+ ///
+ /// Parameters
+ /// ----------
+ /// uri : str
+ /// The URI of the service, including any options as query parameters.
+ /// **kwargs : dict
+ /// Overrides for URI options. Prefer the URI query string.
+ ///
+ /// Returns
+ /// -------
+ /// AsyncOperator
+ /// The new async operator.
+ ///
+ /// Examples
+ /// --------
+ /// ```python
+ /// from urllib.parse import urlencode
+ /// import opendal
+ ///
+ /// op = opendal.AsyncOperator.from_uri("memory://")
+ /// query = urlencode({"region": "us-east-1"})
+ /// op = opendal.AsyncOperator.from_uri(f"s3://bucket/path?{query}")
+ /// ```
+ #[classmethod]
+ #[pyo3(signature = (uri, **kwargs))]
+ pub fn from_uri(
+ _cls: &Bound<PyType>,
+ uri: &str,
+ kwargs: Option<HashMap<String, String>>,
+ ) -> PyResult<Self> {
+ let map = kwargs.unwrap_or_default();
+
+ Ok(AsyncOperator {
+ core: build_operator_from_uri(uri, map.clone())?,
+ __scheme: uri.to_string(),
+ __map: map,
+ })
+ }
+
/// Add a new layer to the operator.
///
/// Parameters
@@ -1716,11 +1846,12 @@ impl AsyncOperator {
)
}
}
- fn __getnewargs_ex__(&self, py: Python) -> PyResult<Py<PyAny>> {
- let args = vec![self.__scheme.clone()];
- let args = PyTuple::new(py, args)?.into_py_any(py)?;
- let kwargs = self.__map.clone().into_py_any(py)?;
- PyTuple::new(py, [args, kwargs])?.into_py_any(py)
+ fn __reduce__(&self, py: Python) -> PyResult<Py<PyAny>> {
+ let reconstructor = py
+ .import("opendal._opendal")?
+ .getattr("_reconstruct_async_operator")?;
+ let args = (self.__scheme.clone(),
self.__map.clone()).into_py_any(py)?;
+ PyTuple::new(py, [reconstructor.into_py_any(py)?,
args])?.into_py_any(py)
}
}
diff --git a/bindings/python/tests/test_from_uri.py
b/bindings/python/tests/test_from_uri.py
new file mode 100644
index 000000000..52e332e85
--- /dev/null
+++ b/bindings/python/tests/test_from_uri.py
@@ -0,0 +1,138 @@
+# 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.
+
+import os
+import pickle
+from random import randint
+from uuid import uuid4
+
+import pytest
+
+import opendal
+from opendal.exceptions import Unsupported
+
+
[email protected](scope="session")
+def uri_async_operator(service_name, setup_config):
+ # Mirror the `async_operator` fixture but construct via `from_uri`: a bare
+ # `scheme://` URI plus the same config passed as keyword options. Apply the
+ # same layers and capability overrides so capability gating and equality
+ # assertions match the constructor-based operator the suite uses.
+ operator = (
+ opendal.AsyncOperator.from_uri(f"{service_name}://", **setup_config)
+ .layer(opendal.layers.RetryLayer())
+ .layer(opendal.layers.ConcurrentLimitLayer(1024))
+ .layer(opendal.layers.MimeGuessLayer())
+ )
+ overrides = os.environ.get("OPENDAL_TEST_CAPABILITY_OVERRIDES")
+ if overrides:
+ operator =
operator.layer(opendal.layers.CapabilityOverrideLayer(overrides))
+ return operator
+
+
[email protected](scope="session")
+def uri_operator(uri_async_operator):
+ return uri_async_operator.to_operator()
+
+
+def test_from_uri_returns_expected_types(uri_operator, uri_async_operator):
+ assert isinstance(uri_operator, opendal.Operator)
+ assert isinstance(uri_async_operator, opendal.AsyncOperator)
+
+
+def test_from_uri_matches_constructor_capability(
+ uri_operator, operator, uri_async_operator, async_operator
+):
+ # `from_uri("scheme://", **config)` and `Operator(scheme, **config)`
resolve
+ # to the same service, so their capabilities must match.
+ assert uri_operator.capability().write == operator.capability().write
+ assert uri_operator.capability().read == operator.capability().read
+ assert uri_async_operator.capability().write ==
async_operator.capability().write
+
+
[email protected]_capability("write", "delete", "stat")
+def test_sync_from_uri_write(service_name, uri_operator):
+ size = randint(1, 1024)
+ filename = f"test_file_{uuid4()}.txt"
+ content = os.urandom(size)
+ uri_operator.write(filename, content)
+ metadata = uri_operator.stat(filename)
+ assert metadata is not None
+ assert metadata.mode.is_file()
+ assert metadata.content_length == size
+
+ uri_operator.delete(filename)
+
+
[email protected]
[email protected]_capability("write", "delete", "stat")
+async def test_async_from_uri_write(service_name, uri_async_operator):
+ size = randint(1, 1024)
+ filename = f"test_file_{uuid4()}.txt"
+ content = os.urandom(size)
+ await uri_async_operator.write(filename, content)
+ metadata = await uri_async_operator.stat(filename)
+ assert metadata is not None
+ assert metadata.mode.is_file()
+ assert metadata.content_length == size
+
+ await uri_async_operator.delete(filename)
+
+
[email protected]_capability("read", "write", "delete")
+def test_sync_from_uri_read_write_roundtrip(uri_operator):
+ # Write then read back through a from_uri-built operator and assert the
+ # content survives the round trip unchanged.
+ size = randint(1, 1024)
+ filename = f"test_file_{uuid4()}.txt"
+ content = os.urandom(size)
+ uri_operator.write(filename, content)
+ assert uri_operator.read(filename) == content
+
+ uri_operator.delete(filename)
+
+
[email protected]
[email protected]_capability("read", "write", "delete")
+async def test_async_from_uri_read_write_roundtrip(uri_async_operator):
+ size = randint(1, 1024)
+ filename = f"test_file_{uuid4()}.txt"
+ content = os.urandom(size)
+ await uri_async_operator.write(filename, content)
+ assert (await uri_async_operator.read(filename)) == content
+
+ await uri_async_operator.delete(filename)
+
+
[email protected]_capability("read", "write", "delete", "shared")
+def test_from_uri_operator_pickle(uri_operator):
+ # Match `test_operator_pickle`: an operator built via `from_uri` must
survive
+ # a pickle round-trip and remain able to read data written before pickling.
+ size = randint(1, 1024)
+ filename = f"random_file_{uuid4()}"
+ content = os.urandom(size)
+ uri_operator.write(filename, content)
+
+ deserialized = pickle.loads(pickle.dumps(uri_operator))
+ assert deserialized.read(filename) == content
+
+ uri_operator.delete(filename)
+
+
+def test_from_uri_unsupported_scheme_raises():
+ with pytest.raises(Unsupported):
+ opendal.Operator.from_uri("thisdoesnotexist://bucket/path")
