| // 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. |
| |
| use std::collections::HashMap; |
| use std::path::PathBuf; |
| use std::time::Duration; |
| |
| use pyo3::IntoPyObjectExt; |
| 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::*; |
| |
| fn build_operator(scheme: &str, map: HashMap<String, String>) -> PyResult<ocore::Operator> { |
| let op = ocore::Operator::via_iter(scheme, map).map_err(format_pyerr)?; |
| Ok(op) |
| } |
| |
| fn build_blocking_operator( |
| scheme: &str, |
| map: HashMap<String, String>, |
| ) -> PyResult<ocore::blocking::Operator> { |
| let op = ocore::Operator::via_iter(scheme, map).map_err(format_pyerr)?; |
| |
| 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 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('_', "-") |
| } |
| |
| /// Convert a config value into the string form core's config deserializer |
| /// consumes. |
| /// |
| /// Accepts the native types the `opendal.config` TypedDicts declare: `str`, |
| /// `bool`, `int`, `os.PathLike`, and `list`/`tuple` of those (`,`-joined, as |
| /// core parses `Vec`). A nested `dict` has no flat-map form and is rejected. |
| fn config_value_to_string(value: &Bound<PyAny>) -> PyResult<String> { |
| // `str` before the list branch (a `str` is also a sequence); `bool` before |
| // `int` (a Python `bool` also extracts as `int`); `dict` before the list |
| // branch so a map is rejected rather than read as its keys. |
| if let Ok(s) = value.extract::<String>() { |
| Ok(s) |
| } else if let Ok(b) = value.extract::<bool>() { |
| Ok(if b { "true" } else { "false" }.to_string()) |
| } else if let Ok(i) = value.extract::<i128>() { |
| Ok(i.to_string()) |
| } else if value.cast::<PyDict>().is_ok() { |
| Err(Unsupported::new_err( |
| "a map-valued config field cannot be built via from_config; leave it unset", |
| )) |
| } else if let Ok(items) = value.extract::<Vec<Bound<PyAny>>>() { |
| let parts = items |
| .iter() |
| .map(config_value_to_string) |
| .collect::<PyResult<Vec<_>>>()?; |
| Ok(parts.join(",")) |
| } else if let Ok(path) = value.extract::<PathBuf>() { |
| Ok(path.to_string_lossy().into_owned()) |
| } else { |
| Err(Unsupported::new_err( |
| "unsupported config value type; pass a str, bool, int, os.PathLike, or list of those", |
| )) |
| } |
| } |
| |
| /// Extract `(scheme, config_map)` from a typed service config dict. |
| /// |
| /// A config is a plain `dict` (an `opendal.config.ServiceConfig`, e.g. |
| /// `S3Config`) whose `scheme` key selects the service; every other pair becomes |
| /// a config option, converted via [`config_value_to_string`]. |
| fn extract_typed_config(config: &Bound<PyAny>) -> PyResult<(String, HashMap<String, String>)> { |
| let dict = config.cast::<PyDict>().map_err(|_| { |
| Unsupported::new_err( |
| "from_config expects an opendal.config.ServiceConfig \ |
| (a dict with a 'scheme' key, e.g. S3Config(scheme=\"s3\", ...))", |
| ) |
| })?; |
| |
| let scheme = dict |
| .get_item("scheme")? |
| .ok_or_else(|| Unsupported::new_err("config is missing the required 'scheme' key"))? |
| .extract::<String>()?; |
| |
| let mut map = HashMap::with_capacity(dict.len()); |
| for (k, v) in dict.iter() { |
| let key = k.extract::<String>()?; |
| if key != "scheme" { |
| let value = config_value_to_string(&v)?; |
| map.insert(key, value); |
| } |
| } |
| |
| Ok((scheme, map)) |
| } |
| |
| /// 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. |
| /// |
| /// See also |
| /// -------- |
| /// AsyncOperator |
| #[pyclass(module = "opendal.operator")] |
| pub struct Operator { |
| core: ocore::blocking::Operator, |
| __scheme: String, |
| __map: HashMap<String, String>, |
| } |
| #[pymethods] |
| impl Operator { |
| /// Create a new blocking `Operator`. |
| /// |
| /// Parameters |
| /// ---------- |
| /// scheme : str | Scheme |
| /// The scheme of the service. |
| /// **kwargs : dict |
| /// The options for the service. |
| /// |
| /// Returns |
| /// ------- |
| /// Operator |
| /// The new operator. |
| #[new] |
| #[pyo3(signature = (scheme: "str | Scheme", *, **kwargs))] |
| 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>() { |
| String::from(py_scheme) |
| } else { |
| return Err(Unsupported::new_err( |
| "Invalid type for scheme, expected str or Scheme", |
| )); |
| }; |
| let scheme = normalize_scheme(&scheme); |
| let map = kwargs.unwrap_or_default(); |
| |
| Ok(Operator { |
| core: build_blocking_operator(&scheme, map.clone())?, |
| __scheme: scheme, |
| __map: map, |
| }) |
| } |
| |
| /// 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, |
| }) |
| } |
| |
| /// Create a new blocking `Operator` from a typed service config. |
| /// |
| /// The config is an ``opendal.config.ServiceConfig`` (e.g. ``S3Config``); its |
| /// ``scheme`` key selects the service, so a static type checker rejects a |
| /// wrong scheme, a missing required key, an unknown key, and a wrong value |
| /// type. Non-string values (``bool``, ``int``, ``os.PathLike``, ``list``) |
| /// are converted to the string form core consumes. |
| /// |
| /// Parameters |
| /// ---------- |
| /// config : ServiceConfig |
| /// A service configuration such as ``opendal.config.S3Config``. |
| /// |
| /// Returns |
| /// ------- |
| /// Operator |
| /// The new operator. |
| /// |
| /// Examples |
| /// -------- |
| /// ```python |
| /// import opendal |
| /// from opendal.config import S3Config |
| /// |
| /// op = opendal.Operator.from_config(S3Config(scheme="s3", bucket="my-bucket")) |
| /// ``` |
| #[classmethod] |
| #[pyo3(signature = (config: "ServiceConfig"))] |
| pub fn from_config(_cls: &Bound<PyType>, config: &Bound<PyAny>) -> PyResult<Self> { |
| let (scheme, map) = extract_typed_config(config)?; |
| let scheme = normalize_scheme(&scheme); |
| |
| Ok(Operator { |
| core: build_blocking_operator(&scheme, map.clone())?, |
| __scheme: scheme, |
| __map: map, |
| }) |
| } |
| |
| /// Add a new layer to this operator. |
| /// |
| /// Parameters |
| /// ---------- |
| /// layer : Layer |
| /// The layer to add. |
| /// |
| /// Returns |
| /// ------- |
| /// Operator |
| /// A new operator with the layer added. |
| pub fn layer(&self, layer: &layers::Layer) -> PyResult<Self> { |
| let op = layer.0.layer(self.core.clone().into()); |
| |
| let runtime = pyo3_async_runtimes::tokio::get_runtime(); |
| let _guard = runtime.enter(); |
| let op = ocore::blocking::Operator::new(op).map_err(format_pyerr)?; |
| Ok(Self { |
| core: op, |
| __scheme: self.__scheme.clone(), |
| __map: self.__map.clone(), |
| }) |
| } |
| |
| /// Open a file-like object for the given path. |
| /// |
| /// The returning file-like object is a context manager. |
| /// |
| /// Parameters |
| /// ---------- |
| /// path : str |
| /// The path to the file. |
| /// mode : str |
| /// The mode to open the file in. Only "rb" and "wb" are supported. |
| /// **kwargs |
| /// Additional options for the underlying reader or writer. |
| /// |
| /// Returns |
| /// ------- |
| /// File |
| /// A file-like object. |
| #[pyo3(signature = (path, mode, *, **kwargs))] |
| pub fn open( |
| &self, |
| path: PathBuf, |
| mode: String, |
| kwargs: Option<&Bound<PyDict>>, |
| ) -> PyResult<File> { |
| let this = self.core.clone(); |
| let path = path.to_string_lossy().to_string(); |
| |
| let reader_opts = kwargs |
| .map(|v| v.extract::<ReadOptions>()) |
| .transpose()? |
| .unwrap_or_default(); |
| |
| let writer_opts = kwargs |
| .map(|v| v.extract::<WriteOptions>()) |
| .transpose()? |
| .unwrap_or_default(); |
| |
| if mode == "rb" { |
| let range = reader_opts.make_range(); |
| let reader = this |
| .reader_options(&path, reader_opts.into()) |
| .map_err(format_pyerr)?; |
| |
| let r = reader |
| .into_std_read(range.to_range()) |
| .map_err(format_pyerr)?; |
| Ok(File::new_reader(r)) |
| } else if mode == "wb" { |
| let writer = this |
| .writer_options(&path, writer_opts.into()) |
| .map_err(format_pyerr)?; |
| Ok(File::new_writer(writer)) |
| } else { |
| Err(Unsupported::new_err(format!( |
| "OpenDAL doesn't support mode: {mode}" |
| ))) |
| } |
| } |
| |
| /// Read the entire contents of a file at the given path. |
| /// |
| /// Parameters |
| /// ---------- |
| /// path : str |
| /// The path to the file. |
| /// version : str, optional |
| /// The version of the file. |
| /// concurrent : int, optional |
| /// The number of concurrent readers. |
| /// chunk : int, optional |
| /// The size of each chunk. |
| /// gap : int, optional |
| /// The gap between each chunk. |
| /// offset : int, optional |
| /// The offset of the file. |
| /// prefetch : int, optional |
| /// The number of bytes to prefetch. |
| /// size : int, optional |
| /// The size of the file. |
| /// if_match : str, optional |
| /// The ETag of the file. |
| /// if_none_match : str, optional |
| /// The ETag of the file. |
| /// if_modified_since : str, optional |
| /// The last modified time of the file. |
| /// if_unmodified_since : str, optional |
| /// The last modified time of the file. |
| /// content_type : str, optional |
| /// The content type of the file. |
| /// cache_control : str, optional |
| /// The cache control of the file. |
| /// content_disposition : str, optional |
| /// The content disposition of the file. |
| /// |
| /// Returns |
| /// ------- |
| /// bytes |
| /// The contents of the file as bytes. |
| #[allow(clippy::too_many_arguments)] |
| #[pyo3(signature = (path, *, |
| version=None, |
| concurrent=None, |
| chunk=None, |
| gap=None, |
| offset=None, |
| prefetch=None, |
| size=None, |
| if_match=None, |
| if_none_match=None, |
| if_modified_since=None, |
| if_unmodified_since=None, |
| content_type=None, |
| cache_control=None, |
| content_disposition=None) -> "bytes")] |
| pub fn read<'p>( |
| &'p self, |
| py: Python<'p>, |
| path: PathBuf, |
| version: Option<String>, |
| concurrent: Option<usize>, |
| chunk: Option<usize>, |
| gap: Option<usize>, |
| offset: Option<usize>, |
| prefetch: Option<usize>, |
| size: Option<usize>, |
| if_match: Option<String>, |
| if_none_match: Option<String>, |
| if_modified_since: Option<jiff::Timestamp>, |
| if_unmodified_since: Option<jiff::Timestamp>, |
| content_type: Option<String>, |
| cache_control: Option<String>, |
| content_disposition: Option<String>, |
| ) -> PyResult<Bound<'p, PyAny>> { |
| let path = path.to_string_lossy().to_string(); |
| let opts = ReadOptions { |
| version, |
| concurrent, |
| chunk, |
| gap, |
| offset, |
| prefetch, |
| size, |
| if_match, |
| if_none_match, |
| if_modified_since, |
| if_unmodified_since, |
| content_type, |
| cache_control, |
| content_disposition, |
| }; |
| let buffer = self |
| .core |
| .read_options(&path, opts.into()) |
| .map_err(format_pyerr)? |
| .to_vec(); |
| |
| Buffer::new(buffer).into_bytes_ref(py) |
| } |
| |
| /// Write bytes to a file at the given path. |
| /// |
| /// This function will create a file if it does not exist, and will |
| /// overwrite its contents if it does. |
| /// |
| /// Parameters |
| /// ---------- |
| /// path : str |
| /// The path to the file. |
| /// bs : bytes |
| /// The contents to write to the file. |
| /// append : bool, optional |
| /// Whether to append to the file instead of overwriting it. |
| /// chunk : int, optional |
| /// The chunk size to use when writing the file. |
| /// concurrent : int, optional |
| /// The number of concurrent requests to make when writing the file. |
| /// cache_control : str, optional |
| /// The cache control header to set on the file. |
| /// content_type : str, optional |
| /// The content type header to set on the file. |
| /// content_disposition : str, optional |
| /// The content disposition header to set on the file. |
| /// content_encoding : str, optional |
| /// The content encoding header to set on the file. |
| /// if_match : str, optional |
| /// The ETag to match when writing the file. |
| /// if_none_match : str, optional |
| /// The ETag to not match when writing the file. |
| /// if_not_exists : bool, optional |
| /// Whether to fail if the file already exists. |
| /// user_metadata : dict, optional |
| /// The user metadata to set on the file. |
| #[allow(clippy::too_many_arguments)] |
| #[pyo3(signature = (path, bs: "bytes", *, |
| append= None, |
| chunk = None, |
| concurrent = None, |
| cache_control = None, |
| content_type = None, |
| content_disposition = None, |
| content_encoding = None, |
| if_match = None, |
| if_none_match = None, |
| if_not_exists = None, |
| user_metadata = None))] |
| pub fn write( |
| &self, |
| path: PathBuf, |
| bs: Vec<u8>, |
| append: Option<bool>, |
| chunk: Option<usize>, |
| concurrent: Option<usize>, |
| cache_control: Option<String>, |
| content_type: Option<String>, |
| content_disposition: Option<String>, |
| content_encoding: Option<String>, |
| if_match: Option<String>, |
| if_none_match: Option<String>, |
| if_not_exists: Option<bool>, |
| user_metadata: Option<HashMap<String, String>>, |
| ) -> PyResult<()> { |
| let path = path.to_string_lossy().to_string(); |
| let opts = WriteOptions { |
| append, |
| chunk, |
| concurrent, |
| cache_control, |
| content_type, |
| content_disposition, |
| content_encoding, |
| if_match, |
| if_none_match, |
| if_not_exists, |
| user_metadata, |
| }; |
| |
| self.core |
| .write_options(&path, bs, opts.into()) |
| .map(|_| ()) |
| .map_err(format_pyerr) |
| } |
| |
| /// Get the metadata of a file at the given path. |
| /// |
| /// Parameters |
| /// ---------- |
| /// path : str |
| /// The path to the file. |
| /// version : str, optional |
| /// The version of the file. |
| /// if_match : str, optional |
| /// The ETag of the file. |
| /// if_none_match : str, optional |
| /// The ETag of the file. |
| /// if_modified_since : datetime, optional |
| /// The last modified time of the file. |
| /// if_unmodified_since : datetime, optional |
| /// The last modified time of the file. |
| /// content_type : str, optional |
| /// The content type of the file. |
| /// cache_control : str, optional |
| /// The cache control of the file. |
| /// content_disposition : str, optional |
| /// The content disposition of the file. |
| /// |
| /// Returns |
| /// ------- |
| /// Metadata |
| /// The metadata of the file. |
| #[allow(clippy::too_many_arguments)] |
| #[pyo3(signature = (path, *, |
| version=None, |
| if_match=None, |
| if_none_match=None, |
| if_modified_since=None, |
| if_unmodified_since=None, |
| content_type=None, |
| cache_control=None, |
| content_disposition=None))] |
| pub fn stat( |
| &self, |
| path: PathBuf, |
| version: Option<String>, |
| if_match: Option<String>, |
| if_none_match: Option<String>, |
| if_modified_since: Option<jiff::Timestamp>, |
| if_unmodified_since: Option<jiff::Timestamp>, |
| content_type: Option<String>, |
| cache_control: Option<String>, |
| content_disposition: Option<String>, |
| ) -> PyResult<Metadata> { |
| let path = path.to_string_lossy().to_string(); |
| let opts = StatOptions { |
| version, |
| if_match, |
| if_none_match, |
| if_modified_since, |
| if_unmodified_since, |
| content_type, |
| cache_control, |
| content_disposition, |
| }; |
| self.core |
| .stat_options(&path, opts.into()) |
| .map_err(format_pyerr) |
| .map(Metadata::new) |
| } |
| |
| /// Copy a file from one path to another. |
| /// |
| /// Parameters |
| /// ---------- |
| /// source : str |
| /// The path to the source file. |
| /// target : str |
| /// The path to the target file. |
| pub fn copy(&self, source: PathBuf, target: PathBuf) -> PyResult<()> { |
| let source = source.to_string_lossy().to_string(); |
| let target = target.to_string_lossy().to_string(); |
| self.core |
| .copy(&source, &target) |
| .map(|_| ()) |
| .map_err(format_pyerr) |
| } |
| |
| /// Rename (move) a file from one path to another. |
| /// |
| /// Parameters |
| /// ---------- |
| /// source : str |
| /// The path to the source file. |
| /// target : str |
| /// The path to the target file. |
| pub fn rename(&self, source: PathBuf, target: PathBuf) -> PyResult<()> { |
| let source = source.to_string_lossy().to_string(); |
| let target = target.to_string_lossy().to_string(); |
| self.core.rename(&source, &target).map_err(format_pyerr) |
| } |
| |
| /// Recursively remove all files and directories at the given path. |
| /// |
| /// Parameters |
| /// ---------- |
| /// path : str |
| /// The path to remove. |
| pub fn remove_all(&self, path: PathBuf) -> PyResult<()> { |
| use ocore::options::DeleteOptions; |
| let path = path.to_string_lossy().to_string(); |
| self.core |
| .delete_options( |
| &path, |
| DeleteOptions { |
| recursive: true, |
| ..Default::default() |
| }, |
| ) |
| .map_err(format_pyerr) |
| } |
| |
| /// Create a directory at the given path. |
| /// |
| /// Notes |
| /// ----- |
| /// To indicate that a path is a directory, it must end with a `/`. |
| /// This operation is always recursive, like `mkdir -p`. |
| /// |
| /// Parameters |
| /// ---------- |
| /// path : str |
| /// The path to the directory. |
| pub fn create_dir(&self, path: PathBuf) -> PyResult<()> { |
| let path = path.to_string_lossy().to_string(); |
| self.core.create_dir(&path).map_err(format_pyerr) |
| } |
| |
| /// Delete a file at the given path. |
| /// |
| /// Notes |
| /// ----- |
| /// This operation will not return an error if the path does not exist. |
| /// |
| /// Parameters |
| /// ---------- |
| /// path : str |
| /// The path to the file. |
| /// 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(); |
| 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. |
| /// |
| /// Parameters |
| /// ---------- |
| /// path : str |
| /// The path to check. |
| /// |
| /// Returns |
| /// ------- |
| /// bool |
| /// True if the path exists, False otherwise. |
| pub fn exists(&self, path: PathBuf) -> PyResult<bool> { |
| let path = path.to_string_lossy().to_string(); |
| self.core.exists(&path).map_err(format_pyerr) |
| } |
| |
| /// List entries in the given directory. |
| /// |
| /// Parameters |
| /// ---------- |
| /// path : str |
| /// The path to the directory. |
| /// limit : int, optional |
| /// The maximum number of entries to return. |
| /// start_after : str, optional |
| /// The entry to start after. |
| /// recursive : bool, optional |
| /// Whether to list recursively. |
| /// versions : bool, optional |
| /// Whether to list versions. |
| /// deleted : bool, optional |
| /// Whether to list deleted entries. |
| /// |
| /// Returns |
| /// ------- |
| /// BlockingLister |
| /// An iterator over the entries in the directory. |
| #[pyo3(signature = (path, *, |
| limit=None, |
| start_after=None, |
| recursive=None, |
| versions=None, |
| deleted=None) -> "collections.abc.Iterable[Entry]")] |
| pub fn list( |
| &self, |
| path: PathBuf, |
| limit: Option<usize>, |
| start_after: Option<String>, |
| recursive: Option<bool>, |
| versions: Option<bool>, |
| deleted: Option<bool>, |
| ) -> PyResult<BlockingLister> { |
| let path = path.to_string_lossy().to_string(); |
| |
| let opts = ListOptions { |
| limit, |
| start_after, |
| recursive, |
| versions, |
| deleted, |
| }; |
| |
| let l = self |
| .core |
| .lister_options(&path, opts.into()) |
| .map_err(format_pyerr)?; |
| Ok(BlockingLister::new(l)) |
| } |
| |
| /// Recursively list entries in the given directory. |
| /// |
| /// Deprecated |
| /// ---------- |
| /// Use `list()` with `recursive=True` instead. |
| /// |
| /// Parameters |
| /// ---------- |
| /// path : str |
| /// The path to the directory. |
| /// limit : int, optional |
| /// The maximum number of entries to return. |
| /// start_after : str, optional |
| /// The entry to start after. |
| /// versions : bool, optional |
| /// Whether to list versions. |
| /// deleted : bool, optional |
| /// Whether to list deleted entries. |
| /// |
| /// Returns |
| /// ------- |
| /// BlockingLister |
| /// An iterator over the entries in the directory. |
| #[pyo3(signature = (path, *, |
| limit=None, |
| start_after=None, |
| versions=None, |
| deleted=None) -> "collections.abc.Iterable[Entry]")] |
| pub fn scan( |
| &self, |
| path: PathBuf, |
| limit: Option<usize>, |
| start_after: Option<String>, |
| versions: Option<bool>, |
| deleted: Option<bool>, |
| ) -> PyResult<BlockingLister> { |
| self.list(path, limit, start_after, Some(true), versions, deleted) |
| } |
| |
| /// Get all capabilities of this operator. |
| /// |
| /// Returns |
| /// ------- |
| /// Capability |
| /// The capability of the operator. |
| pub fn capability(&self) -> PyResult<capability::Capability> { |
| Ok(capability::Capability::new(self.core.info().capability())) |
| } |
| |
| /// Check if the operator is able to work correctly. |
| /// |
| /// Raises |
| /// ------ |
| /// Exception |
| /// If the operator is not able to work correctly. |
| pub fn check(&self) -> PyResult<()> { |
| self.core.check().map_err(format_pyerr) |
| } |
| |
| /// Create a new `AsyncOperator` from this blocking operator. |
| /// |
| /// Returns |
| /// ------- |
| /// AsyncOperator |
| /// The async operator. |
| pub fn to_async_operator(&self) -> PyResult<AsyncOperator> { |
| Ok(AsyncOperator { |
| core: self.core.clone().into(), |
| __scheme: self.__scheme.clone(), |
| __map: self.__map.clone(), |
| }) |
| } |
| |
| fn __repr__(&self) -> String { |
| let info = self.core.info(); |
| let name = info.name(); |
| if name.is_empty() { |
| format!("Operator(\"{}\", root=\"{}\")", info.scheme(), info.root()) |
| } else { |
| format!( |
| "Operator(\"{}\", root=\"{}\", name=\"{name}\")", |
| info.scheme(), |
| info.root() |
| ) |
| } |
| } |
| 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) |
| } |
| } |
| |
| /// The async equivalent of `Operator`. |
| /// |
| /// `AsyncOperator` is the entry point for all async APIs. |
| /// |
| /// See also |
| /// -------- |
| /// Operator |
| #[pyclass(module = "opendal.operator")] |
| pub struct AsyncOperator { |
| core: ocore::Operator, |
| __scheme: String, |
| __map: HashMap<String, String>, |
| } |
| #[pymethods] |
| impl AsyncOperator { |
| /// Create a new `AsyncOperator`. |
| /// |
| /// Parameters |
| /// ---------- |
| /// scheme : str | Scheme |
| /// The scheme of the service. |
| /// **kwargs : dict |
| /// The options for the service. |
| /// |
| /// Returns |
| /// ------- |
| /// AsyncOperator |
| /// The new async operator. |
| #[new] |
| #[pyo3(signature = (scheme: "str | Scheme", * ,**kwargs))] |
| 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>() { |
| String::from(py_scheme) |
| } else { |
| return Err(Unsupported::new_err( |
| "Invalid type for scheme, expected str or Scheme", |
| )); |
| }; |
| let scheme = normalize_scheme(&scheme); |
| |
| let map = kwargs.unwrap_or_default(); |
| |
| Ok(AsyncOperator { |
| core: build_operator(&scheme, map.clone())?, |
| __scheme: scheme, |
| __map: map, |
| }) |
| } |
| |
| /// 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, |
| }) |
| } |
| |
| /// Create a new `AsyncOperator` from a typed service config. |
| /// |
| /// The config is an ``opendal.config.ServiceConfig`` (e.g. ``S3Config``); its |
| /// ``scheme`` key selects the service, so a static type checker rejects a |
| /// wrong scheme, a missing required key, an unknown key, and a wrong value |
| /// type. Non-string values (``bool``, ``int``, ``os.PathLike``, ``list``) |
| /// are converted to the string form core consumes. |
| /// |
| /// Parameters |
| /// ---------- |
| /// config : ServiceConfig |
| /// A service configuration such as ``opendal.config.S3Config``. |
| /// |
| /// Returns |
| /// ------- |
| /// AsyncOperator |
| /// The new async operator. |
| /// |
| /// Examples |
| /// -------- |
| /// ```python |
| /// import opendal |
| /// from opendal.config import S3Config |
| /// |
| /// op = opendal.AsyncOperator.from_config(S3Config(scheme="s3", bucket="my-bucket")) |
| /// ``` |
| #[classmethod] |
| #[pyo3(signature = (config: "ServiceConfig"))] |
| pub fn from_config(_cls: &Bound<PyType>, config: &Bound<PyAny>) -> PyResult<Self> { |
| let (scheme, map) = extract_typed_config(config)?; |
| let scheme = normalize_scheme(&scheme); |
| |
| Ok(AsyncOperator { |
| core: build_operator(&scheme, map.clone())?, |
| __scheme: scheme, |
| __map: map, |
| }) |
| } |
| |
| /// Add a new layer to the operator. |
| /// |
| /// Parameters |
| /// ---------- |
| /// layer : Layer |
| /// The layer to add. |
| /// |
| /// Returns |
| /// ------- |
| /// AsyncOperator |
| /// A new operator with the layer added. |
| pub fn layer(&self, layer: &layers::Layer) -> PyResult<Self> { |
| let op = layer.0.layer(self.core.clone()); |
| Ok(Self { |
| core: op, |
| __scheme: self.__scheme.clone(), |
| __map: self.__map.clone(), |
| }) |
| } |
| |
| /// Open an async file-like object for the given path. |
| /// |
| /// The returning async file-like object is a context manager. |
| /// |
| /// Parameters |
| /// ---------- |
| /// path : str |
| /// The path to the file. |
| /// mode : str |
| /// The mode to open the file in. Only "rb" and "wb" are supported. |
| /// **kwargs : dict |
| /// Additional options for the underlying reader or writer. |
| /// |
| /// Returns |
| /// ------- |
| /// coroutine |
| /// An awaitable that returns a file-like object. |
| #[pyo3(signature = (path, mode, *, **kwargs) -> "collections.abc.Awaitable[AsyncFile]")] |
| pub fn open<'p>( |
| &'p self, |
| py: Python<'p>, |
| path: PathBuf, |
| mode: String, |
| kwargs: Option<&Bound<PyDict>>, |
| ) -> PyResult<Bound<'p, PyAny>> { |
| let this = self.core.clone(); |
| let path = path.to_string_lossy().to_string(); |
| |
| let reader_opts = kwargs |
| .map(|v| v.extract::<ReadOptions>()) |
| .transpose()? |
| .unwrap_or_default(); |
| |
| let writer_opts = kwargs |
| .map(|v| v.extract::<WriteOptions>()) |
| .transpose()? |
| .unwrap_or_default(); |
| |
| future_into_py(py, async move { |
| if mode == "rb" { |
| let range = reader_opts.make_range(); |
| let reader = this |
| .reader_options(&path, reader_opts.into()) |
| .await |
| .map_err(format_pyerr)?; |
| |
| let r = reader |
| .into_futures_async_read(range.to_range()) |
| .await |
| .map_err(format_pyerr)?; |
| Ok(AsyncFile::new_reader(r)) |
| } else if mode == "wb" { |
| let writer = this |
| .writer_options(&path, writer_opts.into()) |
| .await |
| .map_err(format_pyerr)?; |
| let w = writer.into_futures_async_write(); |
| Ok(AsyncFile::new_writer(w)) |
| } else { |
| Err(Unsupported::new_err(format!( |
| "OpenDAL doesn't support mode: {mode}" |
| ))) |
| } |
| }) |
| } |
| |
| /// Read the entire contents of a file at the given path. |
| /// |
| /// Parameters |
| /// ---------- |
| /// path : str |
| /// The path to the file. |
| /// version : str, optional |
| /// The version of the file. |
| /// concurrent : int, optional |
| /// The number of concurrent readers. |
| /// chunk : int, optional |
| /// The size of each chunk. |
| /// gap : int, optional |
| /// The gap between each chunk. |
| /// offset : int, optional |
| /// The offset of the file. |
| /// prefetch : int, optional |
| /// The number of bytes to prefetch. |
| /// size : int, optional |
| /// The size of the file. |
| /// if_match : str, optional |
| /// The ETag of the file. |
| /// if_none_match : str, optional |
| /// The ETag of the file. |
| /// if_modified_since : str, optional |
| /// The last modified time of the file. |
| /// if_unmodified_since : str, optional |
| /// The last modified time of the file. |
| /// content_type : str, optional |
| /// The content type of the file. |
| /// cache_control : str, optional |
| /// The cache control of the file. |
| /// content_disposition : str, optional |
| /// The content disposition of the file. |
| /// |
| /// Returns |
| /// ------- |
| /// coroutine |
| /// An awaitable that returns the contents of the file as bytes. |
| #[allow(clippy::too_many_arguments)] |
| #[pyo3(signature = (path, *, |
| version=None, |
| concurrent=None, |
| chunk=None, |
| gap=None, |
| offset=None, |
| prefetch=None, |
| size=None, |
| if_match=None, |
| if_none_match=None, |
| if_modified_since=None, |
| if_unmodified_since=None, |
| content_type=None, |
| cache_control=None, |
| content_disposition=None) -> "collections.abc.Awaitable[bytes]")] |
| pub fn read<'p>( |
| &'p self, |
| py: Python<'p>, |
| path: PathBuf, |
| version: Option<String>, |
| concurrent: Option<usize>, |
| chunk: Option<usize>, |
| gap: Option<usize>, |
| offset: Option<usize>, |
| prefetch: Option<usize>, |
| size: Option<usize>, |
| if_match: Option<String>, |
| if_none_match: Option<String>, |
| if_modified_since: Option<jiff::Timestamp>, |
| if_unmodified_since: Option<jiff::Timestamp>, |
| content_type: Option<String>, |
| cache_control: Option<String>, |
| content_disposition: Option<String>, |
| ) -> PyResult<Bound<'p, PyAny>> { |
| let this = self.core.clone(); |
| let path = path.to_string_lossy().to_string(); |
| let opts = ReadOptions { |
| version, |
| concurrent, |
| chunk, |
| gap, |
| offset, |
| prefetch, |
| size, |
| if_match, |
| if_none_match, |
| if_modified_since, |
| if_unmodified_since, |
| content_type, |
| cache_control, |
| content_disposition, |
| }; |
| future_into_py(py, async move { |
| let range = opts.make_range(); |
| let res = this |
| .reader_options(&path, opts.into()) |
| .await |
| .map_err(format_pyerr)? |
| .read(range.to_range()) |
| .await |
| .map_err(format_pyerr)? |
| .to_vec(); |
| Python::attach(|py| Buffer::new(res).into_bytes(py)) |
| }) |
| } |
| |
| /// Write bytes to a file at the given path. |
| /// |
| /// This function will create a file if it does not exist, and will |
| /// overwrite its contents if it does. |
| /// |
| /// Parameters |
| /// ---------- |
| /// path : str |
| /// The path to the file. |
| /// bs : bytes |
| /// The contents to write to the file. |
| /// append : bool, optional |
| /// Whether to append to the file instead of overwriting it. |
| /// chunk : int, optional |
| /// The chunk size to use when writing the file. |
| /// concurrent : int, optional |
| /// The number of concurrent requests to make when writing the file. |
| /// cache_control : str, optional |
| /// The cache control header to set on the file. |
| /// content_type : str, optional |
| /// The content type header to set on the file. |
| /// content_disposition : str, optional |
| /// The content disposition header to set on the file. |
| /// content_encoding : str, optional |
| /// The content encoding header to set on the file. |
| /// if_match : str, optional |
| /// The ETag to match when writing the file. |
| /// if_none_match : str, optional |
| /// The ETag to not match when writing the file. |
| /// if_not_exists : bool, optional |
| /// Whether to fail if the file already exists. |
| /// user_metadata : dict, optional |
| /// The user metadata to set on the file. |
| /// |
| /// Returns |
| /// ------- |
| /// coroutine |
| /// An awaitable that completes when the write is finished. |
| #[allow(clippy::too_many_arguments)] |
| #[pyo3(signature = (path, bs: "bytes", *, |
| append= None, |
| chunk = None, |
| concurrent = None, |
| cache_control = None, |
| content_type = None, |
| content_disposition = None, |
| content_encoding = None, |
| if_match = None, |
| if_none_match = None, |
| if_not_exists = None, |
| user_metadata = None) -> "collections.abc.Awaitable[None]")] |
| pub fn write<'p>( |
| &'p self, |
| py: Python<'p>, |
| path: PathBuf, |
| bs: &Bound<PyBytes>, |
| append: Option<bool>, |
| chunk: Option<usize>, |
| concurrent: Option<usize>, |
| cache_control: Option<String>, |
| content_type: Option<String>, |
| content_disposition: Option<String>, |
| content_encoding: Option<String>, |
| if_match: Option<String>, |
| if_none_match: Option<String>, |
| if_not_exists: Option<bool>, |
| user_metadata: Option<HashMap<String, String>>, |
| ) -> PyResult<Bound<'p, PyAny>> { |
| let opts = WriteOptions { |
| append, |
| chunk, |
| concurrent, |
| cache_control, |
| content_type, |
| content_disposition, |
| content_encoding, |
| if_match, |
| if_none_match, |
| if_not_exists, |
| user_metadata, |
| }; |
| let this = self.core.clone(); |
| let bs = bs.as_bytes().to_vec(); |
| let path = path.to_string_lossy().to_string(); |
| future_into_py(py, async move { |
| this.write_options(&path, bs, opts.into()) |
| .await |
| .map(|_| ()) |
| .map_err(format_pyerr) |
| }) |
| } |
| |
| /// Get the metadata of a file at the given path. |
| /// |
| /// Parameters |
| /// ---------- |
| /// path : str |
| /// The path to the file. |
| /// version : str, optional |
| /// The version of the file. |
| /// if_match : str, optional |
| /// The ETag of the file. |
| /// if_none_match : str, optional |
| /// The ETag of the file. |
| /// if_modified_since : datetime, optional |
| /// The last modified time of the file. |
| /// if_unmodified_since : datetime, optional |
| /// The last modified time of the file. |
| /// content_type : str, optional |
| /// The content type of the file. |
| /// cache_control : str, optional |
| /// The cache control of the file. |
| /// content_disposition : str, optional |
| /// The content disposition of the file. |
| /// |
| /// Returns |
| /// ------- |
| /// coroutine |
| /// An awaitable that returns the metadata of the file. |
| #[allow(clippy::too_many_arguments)] |
| #[pyo3(signature = (path, *, |
| version=None, |
| if_match=None, |
| if_none_match=None, |
| if_modified_since=None, |
| if_unmodified_since=None, |
| content_type=None, |
| cache_control=None, |
| content_disposition=None) -> "collections.abc.Awaitable[Metadata]")] |
| pub fn stat<'p>( |
| &'p self, |
| py: Python<'p>, |
| path: PathBuf, |
| version: Option<String>, |
| if_match: Option<String>, |
| if_none_match: Option<String>, |
| if_modified_since: Option<jiff::Timestamp>, |
| if_unmodified_since: Option<jiff::Timestamp>, |
| content_type: Option<String>, |
| cache_control: Option<String>, |
| content_disposition: Option<String>, |
| ) -> PyResult<Bound<'p, PyAny>> { |
| let this = self.core.clone(); |
| let path = path.to_string_lossy().to_string(); |
| let opts = StatOptions { |
| version, |
| if_match, |
| if_none_match, |
| if_modified_since, |
| if_unmodified_since, |
| content_type, |
| cache_control, |
| content_disposition, |
| }; |
| |
| future_into_py(py, async move { |
| let res: Metadata = this |
| .stat_options(&path, opts.into()) |
| .await |
| .map_err(format_pyerr) |
| .map(Metadata::new)?; |
| |
| Ok(res) |
| }) |
| } |
| |
| /// Copy a file from one path to another. |
| /// |
| /// Parameters |
| /// ---------- |
| /// source : str |
| /// The path to the source file. |
| /// target : str |
| /// The path to the target file. |
| /// |
| /// Returns |
| /// ------- |
| /// coroutine |
| /// An awaitable that completes when the copy is finished. |
| #[pyo3(signature = (source, target) -> "collections.abc.Awaitable[None]")] |
| pub fn copy<'p>( |
| &'p self, |
| py: Python<'p>, |
| source: PathBuf, |
| target: PathBuf, |
| ) -> PyResult<Bound<'p, PyAny>> { |
| let this = self.core.clone(); |
| let source = source.to_string_lossy().to_string(); |
| let target = target.to_string_lossy().to_string(); |
| future_into_py(py, async move { |
| this.copy(&source, &target) |
| .await |
| .map(|_| ()) |
| .map_err(format_pyerr) |
| }) |
| } |
| |
| /// Rename (move) a file from one path to another. |
| /// |
| /// Parameters |
| /// ---------- |
| /// source : str |
| /// The path to the source file. |
| /// target : str |
| /// The path to the target file. |
| /// |
| /// Returns |
| /// ------- |
| /// coroutine |
| /// An awaitable that completes when the rename is finished. |
| #[pyo3(signature = (source, target) -> "collections.abc.Awaitable[None]")] |
| pub fn rename<'p>( |
| &'p self, |
| py: Python<'p>, |
| source: PathBuf, |
| target: PathBuf, |
| ) -> PyResult<Bound<'p, PyAny>> { |
| let this = self.core.clone(); |
| let source = source.to_string_lossy().to_string(); |
| let target = target.to_string_lossy().to_string(); |
| future_into_py(py, async move { |
| this.rename(&source, &target).await.map_err(format_pyerr) |
| }) |
| } |
| |
| /// Recursively remove all files and directories at the given path. |
| /// |
| /// Parameters |
| /// ---------- |
| /// path : str |
| /// The path to remove. |
| /// |
| /// Returns |
| /// ------- |
| /// coroutine |
| /// An awaitable that completes when the removal is finished. |
| #[pyo3(signature = (path) -> "collections.abc.Awaitable[None]")] |
| pub fn remove_all<'p>(&'p self, py: Python<'p>, path: PathBuf) -> 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_with(&path) |
| .recursive(true) |
| .await |
| .map_err(format_pyerr) |
| }) |
| } |
| |
| /// Check if the operator is able to work correctly. |
| /// |
| /// Returns |
| /// ------- |
| /// coroutine |
| /// An awaitable that completes when the check is finished. |
| /// |
| /// Raises |
| /// ------ |
| /// Exception |
| /// If the operator is not able to work correctly. |
| #[pyo3(signature = () -> "collections.abc.Awaitable[None]")] |
| pub fn check<'p>(&'p self, py: Python<'p>) -> PyResult<Bound<'p, PyAny>> { |
| let this = self.core.clone(); |
| future_into_py(py, async move { this.check().await.map_err(format_pyerr) }) |
| } |
| |
| /// Create a directory at the given path. |
| /// |
| /// Notes |
| /// ----- |
| /// To indicate that a path is a directory, it must end with a `/`. |
| /// This operation is always recursive, like `mkdir -p`. |
| /// |
| /// Parameters |
| /// ---------- |
| /// path : str |
| /// The path to the directory. |
| /// |
| /// Returns |
| /// ------- |
| /// coroutine |
| /// An awaitable that completes when the directory is created. |
| #[pyo3(signature = (path) -> "collections.abc.Awaitable[None]")] |
| pub fn create_dir<'p>(&'p self, py: Python<'p>, path: PathBuf) -> PyResult<Bound<'p, PyAny>> { |
| let this = self.core.clone(); |
| let path = path.to_string_lossy().to_string(); |
| future_into_py(py, async move { |
| this.create_dir(&path).await.map_err(format_pyerr) |
| }) |
| } |
| |
| /// Delete a file at the given path. |
| /// |
| /// Notes |
| /// ----- |
| /// This operation will not return an error if the path does not exist. |
| /// |
| /// Parameters |
| /// ---------- |
| /// path : str |
| /// The path to the file. |
| /// |
| /// Returns |
| /// ------- |
| /// coroutine |
| /// An awaitable that completes when the file is deleted. |
| /// 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) -> "collections.abc.Awaitable[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 { |
| 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. |
| /// |
| /// Parameters |
| /// ---------- |
| /// path : str |
| /// The path to check. |
| /// |
| /// Returns |
| /// ------- |
| /// coroutine |
| /// An awaitable that returns True if the path exists, False otherwise. |
| #[pyo3(signature = (path) -> "collections.abc.Awaitable[bool]")] |
| pub fn exists<'p>(&'p self, py: Python<'p>, path: PathBuf) -> PyResult<Bound<'p, PyAny>> { |
| let this = self.core.clone(); |
| let path = path.to_string_lossy().to_string(); |
| future_into_py( |
| py, |
| async move { this.exists(&path).await.map_err(format_pyerr) }, |
| ) |
| } |
| |
| /// List entries in the given directory. |
| /// |
| /// Parameters |
| /// ---------- |
| /// path : str |
| /// The path to the directory. |
| /// limit : int, optional |
| /// The maximum number of entries to return. |
| /// start_after : str, optional |
| /// The entry to start after. |
| /// recursive : bool, optional |
| /// Whether to list recursively. |
| /// versions : bool, optional |
| /// Whether to list versions. |
| /// deleted : bool, optional |
| /// Whether to list deleted entries. |
| /// |
| /// Returns |
| /// ------- |
| /// coroutine |
| /// An awaitable that returns an async iterator over the entries. |
| #[allow(clippy::too_many_arguments)] |
| #[pyo3(signature = (path, *, |
| limit=None, |
| start_after=None, |
| recursive=None, |
| versions=None, |
| deleted=None) -> "collections.abc.Awaitable[collections.abc.AsyncIterable[Entry]]")] |
| pub fn list<'p>( |
| &'p self, |
| py: Python<'p>, |
| path: PathBuf, |
| limit: Option<usize>, |
| start_after: Option<String>, |
| recursive: Option<bool>, |
| versions: Option<bool>, |
| deleted: Option<bool>, |
| ) -> PyResult<Bound<'p, PyAny>> { |
| let this = self.core.clone(); |
| let path = path.to_string_lossy().to_string(); |
| let opts = ListOptions { |
| limit, |
| start_after, |
| recursive, |
| versions, |
| deleted, |
| }; |
| |
| future_into_py(py, async move { |
| let lister = this |
| .lister_options(&path, opts.into()) |
| .await |
| .map_err(format_pyerr)?; |
| let pylister = Python::attach(|py| AsyncLister::new(lister).into_py_any(py))?; |
| |
| Ok(pylister) |
| }) |
| } |
| |
| /// Recursively list entries in the given directory. |
| /// |
| /// Deprecated |
| /// ---------- |
| /// Use `list()` with `recursive=True` instead. |
| /// |
| /// Parameters |
| /// ---------- |
| /// path : str |
| /// The path to the directory. |
| /// limit : int, optional |
| /// The maximum number of entries to return. |
| /// start_after : str, optional |
| /// The entry to start after. |
| /// versions : bool, optional |
| /// Whether to list versions. |
| /// deleted : bool, optional |
| /// Whether to list deleted entries. |
| /// |
| /// Returns |
| /// ------- |
| /// coroutine |
| /// An awaitable that returns an async iterator over the entries. |
| #[pyo3(signature = (path, *, |
| limit=None, |
| start_after=None, |
| versions=None, |
| deleted=None) -> "collections.abc.Awaitable[collections.abc.AsyncIterable[Entry]]")] |
| pub fn scan<'p>( |
| &'p self, |
| py: Python<'p>, |
| path: PathBuf, |
| limit: Option<usize>, |
| start_after: Option<String>, |
| versions: Option<bool>, |
| deleted: Option<bool>, |
| ) -> PyResult<Bound<'p, PyAny>> { |
| self.list(py, path, limit, start_after, Some(true), versions, deleted) |
| } |
| |
| /// Create a presigned request for a stat operation. |
| /// |
| /// Parameters |
| /// ---------- |
| /// path : str |
| /// The path of the object to stat. |
| /// expire_second : int |
| /// The number of seconds until the presigned URL expires. |
| /// version : str, optional |
| /// The version of the file. |
| /// if_match : str, optional |
| /// The ETag to match. |
| /// if_none_match : str, optional |
| /// The ETag to not match. |
| /// if_modified_since : datetime, optional |
| /// Only return if modified since this time. |
| /// if_unmodified_since : datetime, optional |
| /// Only return if unmodified since this time. |
| /// content_type : str, optional |
| /// Override the content type in the presigned response. |
| /// cache_control : str, optional |
| /// Override the cache control in the presigned response. |
| /// content_disposition : str, optional |
| /// Override the content disposition in the presigned response. |
| /// |
| /// Returns |
| /// ------- |
| /// coroutine |
| /// An awaitable that returns a presigned request object. |
| #[allow(clippy::too_many_arguments)] |
| #[pyo3(signature = (path, expire_second, *, |
| version=None, |
| if_match=None, |
| if_none_match=None, |
| if_modified_since=None, |
| if_unmodified_since=None, |
| content_type=None, |
| cache_control=None, |
| content_disposition=None) -> "collections.abc.Awaitable[PresignedRequest]")] |
| pub fn presign_stat<'p>( |
| &'p self, |
| py: Python<'p>, |
| path: PathBuf, |
| expire_second: u64, |
| version: Option<String>, |
| if_match: Option<String>, |
| if_none_match: Option<String>, |
| if_modified_since: Option<jiff::Timestamp>, |
| if_unmodified_since: Option<jiff::Timestamp>, |
| content_type: Option<String>, |
| cache_control: Option<String>, |
| content_disposition: Option<String>, |
| ) -> PyResult<Bound<'p, PyAny>> { |
| let this = self.core.clone(); |
| let path = path.to_string_lossy().to_string(); |
| let opts = StatOptions { |
| version, |
| if_match, |
| if_none_match, |
| if_modified_since, |
| if_unmodified_since, |
| content_type, |
| cache_control, |
| content_disposition, |
| }; |
| future_into_py(py, async move { |
| let res = this |
| .presign_stat_options(&path, Duration::from_secs(expire_second), opts.into()) |
| .await |
| .map_err(format_pyerr) |
| .map(PresignedRequest)?; |
| |
| Ok(res) |
| }) |
| } |
| |
| /// Create a presigned request for a read operation. |
| /// |
| /// Parameters |
| /// ---------- |
| /// path : str |
| /// The path of the object to read. |
| /// expire_second : int |
| /// The number of seconds until the presigned URL expires. |
| /// version : str, optional |
| /// The version of the file. |
| /// if_match : str, optional |
| /// The ETag to match. |
| /// if_none_match : str, optional |
| /// The ETag to not match. |
| /// if_modified_since : datetime, optional |
| /// Only return if modified since this time. |
| /// if_unmodified_since : datetime, optional |
| /// Only return if unmodified since this time. |
| /// content_type : str, optional |
| /// Override the content type in the presigned response. |
| /// cache_control : str, optional |
| /// Override the cache control in the presigned response. |
| /// content_disposition : str, optional |
| /// Override the content disposition in the presigned response. |
| /// |
| /// Returns |
| /// ------- |
| /// coroutine |
| /// An awaitable that returns a presigned request object. |
| #[allow(clippy::too_many_arguments)] |
| #[pyo3(signature = (path, expire_second, *, |
| version=None, |
| if_match=None, |
| if_none_match=None, |
| if_modified_since=None, |
| if_unmodified_since=None, |
| content_type=None, |
| cache_control=None, |
| content_disposition=None) -> "collections.abc.Awaitable[PresignedRequest]")] |
| pub fn presign_read<'p>( |
| &'p self, |
| py: Python<'p>, |
| path: PathBuf, |
| expire_second: u64, |
| version: Option<String>, |
| if_match: Option<String>, |
| if_none_match: Option<String>, |
| if_modified_since: Option<jiff::Timestamp>, |
| if_unmodified_since: Option<jiff::Timestamp>, |
| content_type: Option<String>, |
| cache_control: Option<String>, |
| content_disposition: Option<String>, |
| ) -> PyResult<Bound<'p, PyAny>> { |
| let this = self.core.clone(); |
| let path = path.to_string_lossy().to_string(); |
| let opts = ReadOptions { |
| version, |
| if_match, |
| if_none_match, |
| if_modified_since, |
| if_unmodified_since, |
| content_type, |
| cache_control, |
| content_disposition, |
| ..Default::default() |
| }; |
| future_into_py(py, async move { |
| let res = this |
| .presign_read_options(&path, Duration::from_secs(expire_second), opts.into()) |
| .await |
| .map_err(format_pyerr) |
| .map(PresignedRequest)?; |
| |
| Ok(res) |
| }) |
| } |
| |
| /// Create a presigned request for a write operation. |
| /// |
| /// Parameters |
| /// ---------- |
| /// path : str |
| /// The path of the object to write to. |
| /// expire_second : int |
| /// The number of seconds until the presigned URL expires. |
| /// content_type : str, optional |
| /// The content type header to set on the file. |
| /// content_disposition : str, optional |
| /// The content disposition header to set on the file. |
| /// content_encoding : str, optional |
| /// The content encoding header to set on the file. |
| /// cache_control : str, optional |
| /// The cache control header to set on the file. |
| /// if_match : str, optional |
| /// The ETag to match when writing the file. |
| /// if_none_match : str, optional |
| /// The ETag to not match when writing the file. |
| /// if_not_exists : bool, optional |
| /// Whether to fail if the file already exists. |
| /// user_metadata : dict, optional |
| /// The user metadata to set on the file. |
| /// |
| /// Returns |
| /// ------- |
| /// coroutine |
| /// An awaitable that returns a presigned request object. |
| #[allow(clippy::too_many_arguments)] |
| #[pyo3(signature = (path, expire_second, *, |
| content_type=None, |
| content_disposition=None, |
| content_encoding=None, |
| cache_control=None, |
| if_match=None, |
| if_none_match=None, |
| if_not_exists=None, |
| user_metadata=None) -> "collections.abc.Awaitable[PresignedRequest]")] |
| pub fn presign_write<'p>( |
| &'p self, |
| py: Python<'p>, |
| path: PathBuf, |
| expire_second: u64, |
| content_type: Option<String>, |
| content_disposition: Option<String>, |
| content_encoding: Option<String>, |
| cache_control: Option<String>, |
| if_match: Option<String>, |
| if_none_match: Option<String>, |
| if_not_exists: Option<bool>, |
| user_metadata: Option<HashMap<String, String>>, |
| ) -> PyResult<Bound<'p, PyAny>> { |
| let this = self.core.clone(); |
| let path = path.to_string_lossy().to_string(); |
| let opts = WriteOptions { |
| content_type, |
| content_disposition, |
| content_encoding, |
| cache_control, |
| if_match, |
| if_none_match, |
| if_not_exists, |
| user_metadata, |
| ..Default::default() |
| }; |
| future_into_py(py, async move { |
| let res = this |
| .presign_write_options(&path, Duration::from_secs(expire_second), opts.into()) |
| .await |
| .map_err(format_pyerr) |
| .map(PresignedRequest)?; |
| |
| Ok(res) |
| }) |
| } |
| |
| /// Create a presigned request for a delete operation. |
| /// |
| /// Parameters |
| /// ---------- |
| /// path : str |
| /// The path of the object to delete. |
| /// expire_second : int |
| /// The number of seconds until the presigned URL expires. |
| /// version : str, optional |
| /// The version of the file to delete. |
| /// |
| /// Returns |
| /// ------- |
| /// coroutine |
| /// An awaitable that returns a presigned request object. |
| #[pyo3(signature = (path, expire_second, *, version=None) -> "collections.abc.Awaitable[PresignedRequest]")] |
| pub fn presign_delete<'p>( |
| &'p self, |
| py: Python<'p>, |
| path: PathBuf, |
| expire_second: u64, |
| version: Option<String>, |
| ) -> PyResult<Bound<'p, PyAny>> { |
| let this = self.core.clone(); |
| let path = path.to_string_lossy().to_string(); |
| 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()) |
| .await |
| .map_err(format_pyerr) |
| .map(PresignedRequest)?; |
| |
| Ok(res) |
| }) |
| } |
| |
| /// Get all capabilities of this operator. |
| /// |
| /// Returns |
| /// ------- |
| /// Capability |
| /// The capability of the operator. |
| pub fn capability(&self) -> PyResult<Capability> { |
| Ok(capability::Capability::new(self.core.info().capability())) |
| } |
| |
| /// Create a new blocking `Operator` from this async operator. |
| /// |
| /// Returns |
| /// ------- |
| /// Operator |
| /// The blocking operator. |
| pub fn to_operator(&self) -> PyResult<Operator> { |
| let runtime = pyo3_async_runtimes::tokio::get_runtime(); |
| let _guard = runtime.enter(); |
| let op = ocore::blocking::Operator::new(self.core.clone()).map_err(format_pyerr)?; |
| |
| Ok(Operator { |
| core: op, |
| __scheme: self.__scheme.clone(), |
| __map: self.__map.clone(), |
| }) |
| } |
| |
| fn __repr__(&self) -> String { |
| let info = self.core.info(); |
| let name = info.name(); |
| if name.is_empty() { |
| format!( |
| "AsyncOperator(\"{}\", root=\"{}\")", |
| info.scheme(), |
| info.root() |
| ) |
| } else { |
| format!( |
| "AsyncOperator(\"{}\", root=\"{}\", name=\"{name}\")", |
| info.scheme(), |
| info.root() |
| ) |
| } |
| } |
| 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) |
| } |
| } |
| |
| /// A presigned request. |
| /// |
| /// This contains the information required to make a request to the |
| /// underlying service, including the URL, method, and headers. |
| #[pyclass(module = "opendal.types")] |
| pub struct PresignedRequest(ocore::raw::PresignedRequest); |
| #[pymethods] |
| impl PresignedRequest { |
| /// The URL of this request. |
| #[getter] |
| pub fn url(&self) -> String { |
| self.0.uri().to_string() |
| } |
| |
| /// The HTTP method of this request. |
| #[getter] |
| pub fn method(&self) -> &str { |
| self.0.method().as_str() |
| } |
| |
| /// The HTTP headers of this request. |
| /// |
| /// Returns |
| /// ------- |
| /// dict |
| /// The HTTP headers of this request. |
| #[getter] |
| pub fn headers(&self) -> PyResult<HashMap<&str, &str>> { |
| let mut headers = HashMap::new(); |
| for (k, v) in self.0.header().iter() { |
| let k = k.as_str(); |
| let v = v |
| .to_str() |
| .map_err(|err| Unexpected::new_err(err.to_string()))?; |
| if headers.insert(k, v).is_some() { |
| return Err(Unexpected::new_err("duplicate header")); |
| } |
| } |
| Ok(headers) |
| } |
| } |
