feat: initial Operator::from_uri implementation

core/src/raw/http_util/mod.rs

@@ -57,6 +57,7 @@ mod uri;
 pub use uri::new_http_uri_invalid_error;
 pub use uri::percent_decode_path;
 pub use uri::percent_encode_path;
+pub use uri::query_pairs;
 pub use uri::QueryPairsWriter;
 
 mod error;

core/src/raw/http_util/uri.rs

@@ -64,6 +64,28 @@ pub fn percent_decode_path(path: &str) -> String {
     }
 }
 
+/// query_pairs will parse a query string encoded as key-value pairs separated by `&` to a vector of key-value pairs.
+/// It also does percent decoding for both key and value.
+///
+/// Note that `?` is not allowed in the query string, and it will be treated as a part of the first key if included.
+pub fn query_pairs(query: &str) -> Vec<(String, String)> {
+    query
+        .split('&')
+        .filter_map(|pair| {
+            let mut iter = pair.splitn(2, '=');
+
+            let key = iter.next()?;
+            if key.is_empty() {
+                return None;
+            }
+
+            let value = iter.next().unwrap_or("");
+            Some((key, value))
+        })
+        .map(|(key, value)| (percent_decode_path(key), percent_decode_path(value)))
+        .collect()
+}
+
 /// QueryPairsWriter is used to write query pairs to a url.
 pub struct QueryPairsWriter {
     base: String,
@@ -186,4 +208,37 @@ mod tests {
             assert_eq!(actual, expected, "{name}");
         }
     }
+
+    #[test]
+    fn test_query_pairs() {
+        let cases = vec![
+            (
+                "single pair",
+                "key=value",
+                vec![("key".into(), "value".into())],
+            ),
+            (
+                "multiple pairs",
+                "key=value&key2=value2&key3=value3",
+                vec![
+                    ("key".into(), "value".into()),
+                    ("key2".into(), "value2".into()),
+                    ("key3".into(), "value3".into()),
+                ],
+            ),
+            ("empty value", "key=", vec![("key".into(), "".into())]),
+            ("empty input", "", vec![]),
+            (
+                "Unicode Characters",
+                "unicode%20param=%E4%BD%A0%E5%A5%BD%EF%BC%8C%E4%B8%96%E7%95%8C%EF%BC%81%E2%9D%A4",
+                vec![("unicode param".into(), "你好，世界！❤".into())],
+            ),
+        ];
+
+        for (name, input, expected) in cases {
+            let actual = query_pairs(input);
+
+            assert_eq!(actual, expected, "{name}");
+        }
+    }
 }

core/src/types/builder.rs

@@ -17,11 +17,13 @@
 
 use std::fmt::Debug;
 
+use http::Uri;
 use serde::de::DeserializeOwned;
 use serde::Serialize;
 
 use crate::raw::*;
 use crate::*;
+use types::operator::{OperatorFactory, OperatorRegistry};
 
 /// Builder is used to set up underlying services.
 ///
@@ -56,6 +58,15 @@ pub trait Builder: Default + 'static {
 
     /// Consume the accessor builder to build a service.
     fn build(self) -> Result<impl Access>;
+
+    /// Register this builder in the given registry.
+    fn register_in_registry(registry: &mut OperatorRegistry) {
+        let operator_factory: OperatorFactory = |uri, options| {
+            let builder = Self::Config::from_uri(uri, options)?.into_builder();
+            Ok(Operator::new(builder)?.finish())
+        };
+        registry.register(Self::SCHEME.into_static(), operator_factory)
+    }
 }
 
 /// Dummy implementation of builder
@@ -137,6 +148,15 @@ pub trait Configurator: Serialize + DeserializeOwned + Debug + 'static {
         })
     }
 
+    /// TODO: document this.
+    fn from_uri(uri: &Uri, options: impl IntoIterator<Item = (String, String)>) -> Result<Self> {
+        let query_pairs = uri.query().map(query_pairs).unwrap_or_default();
+
+        let merged_options = query_pairs.into_iter().chain(options);
+
+        Self::from_iter(merged_options)
+    }
+
     /// Convert this configuration into a service builder.
     fn into_builder(self) -> Self::Builder;
 }

core/src/types/mod.rs

@@ -55,7 +55,10 @@ pub use operator::operator_futures;
 pub use operator::BlockingOperator;
 pub use operator::Operator;
 pub use operator::OperatorBuilder;
+pub use operator::OperatorFactory;
 pub use operator::OperatorInfo;
+pub use operator::OperatorRegistry;
+pub use operator::GLOBAL_OPERATOR_REGISTRY;
 
 mod builder;
 pub use builder::Builder;

core/src/types/operator/builder.rs

@@ -18,6 +18,7 @@
 use std::collections::HashMap;
 use std::sync::Arc;
 
+use super::registry::GLOBAL_OPERATOR_REGISTRY;
 use crate::layers::*;
 use crate::raw::*;
 use crate::*;
@@ -95,6 +96,26 @@ impl Operator {
         Ok(OperatorBuilder::new(acc))
     }
 
+    /// TODO: document this.
+    ///
+    /// TODO: improve those examples
+    /// # Examples
+    /// ```
+    /// # use anyhow::Result;
+    /// use opendal::Operator;
+    ///
+    /// fn test() -> Result<()> {
+    ///     Operator::from_uri("fs://?root=/tmp/test", vec![])?;
+    ///      Ok(())
+    /// }
+    /// ```
+    pub fn from_uri(
+        uri: &str,
+        options: impl IntoIterator<Item = (String, String)>,
+    ) -> Result<Self> {
+        GLOBAL_OPERATOR_REGISTRY.parse(uri, options)
+    }
+
     /// Create a new operator from given iterator in static dispatch.
     ///
     /// # Notes

core/src/types/operator/mod.rs

@@ -32,3 +32,8 @@ pub use info::OperatorInfo;
 
 pub mod operator_functions;
 pub mod operator_futures;
+
+mod registry;
+pub use registry::OperatorFactory;
+pub use registry::OperatorRegistry;
+pub use registry::GLOBAL_OPERATOR_REGISTRY;

core/src/types/operator/registry.rs

@@ -0,0 +1,204 @@
+// 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::sync::LazyLock;
+
+use http::Uri;
+
+use crate::services::*;
+use crate::*;
+
+/// TODO: document this
+pub static GLOBAL_OPERATOR_REGISTRY: LazyLock<OperatorRegistry> =
+    LazyLock::new(OperatorRegistry::initialized);
+
+/// TODO: document this
+pub type OperatorFactory = fn(&Uri, HashMap<String, String>) -> Result<Operator>;
+
+/// TODO: document this
+#[derive(Clone, Debug, Default)]
+pub struct OperatorRegistry {
+    registry: HashMap<String, OperatorFactory>,
+}
+
+impl OperatorRegistry {
+    /// TODO: document this
+    pub fn new() -> Self {
+        Self {
+            registry: HashMap::new(),
+        }
+    }
+    /// TODO: document this
+    pub fn initialized() -> Self {
+        let mut registry = Self::new();
+
+        #[cfg(feature = "services-aliyun-drive")]
+        AliyunDrive::register_in_registry(&mut registry);
+        #[cfg(feature = "services-atomicserver")]
+        Atomicserver::register_in_registry(&mut registry);
+        #[cfg(feature = "services-alluxio")]
+        Alluxio::register_in_registry(&mut registry);
+        #[cfg(feature = "services-azblob")]
+        Azblob::register_in_registry(&mut registry);
+        #[cfg(feature = "services-azdls")]
+        Azdls::register_in_registry(&mut registry);
+        #[cfg(feature = "services-azfile")]
+        Azfile::register_in_registry(&mut registry);
+        #[cfg(feature = "services-b2")]
+        B2::register_in_registry(&mut registry);
+        #[cfg(feature = "services-cacache")]
+        Cacache::register_in_registry(&mut registry);
+        #[cfg(feature = "services-cos")]
+        Cos::register_in_registry(&mut registry);
+        #[cfg(feature = "services-compfs")]
+        Compfs::register_in_registry(&mut registry);
+        #[cfg(feature = "services-dashmap")]
+        Dashmap::register_in_registry(&mut registry);
+        #[cfg(feature = "services-dropbox")]
+        Dropbox::register_in_registry(&mut registry);
+        #[cfg(feature = "services-etcd")]
+        Etcd::register_in_registry(&mut registry);
+        #[cfg(feature = "services-foundationdb")]
+        Foundationdb::register_in_registry(&mut registry);
+        #[cfg(feature = "services-fs")]
+        Fs::register_in_registry(&mut registry);
+        #[cfg(feature = "services-ftp")]
+        Ftp::register_in_registry(&mut registry);
+        #[cfg(feature = "services-gcs")]
+        Gcs::register_in_registry(&mut registry);
+        #[cfg(feature = "services-ghac")]
+        Ghac::register_in_registry(&mut registry);
+        #[cfg(feature = "services-hdfs")]
+        Hdfs::register_in_registry(&mut registry);
+        #[cfg(feature = "services-http")]
+        Http::register_in_registry(&mut registry);
+        #[cfg(feature = "services-huggingface")]
+        Huggingface::register_in_registry(&mut registry);
+        #[cfg(feature = "services-ipfs")]
+        Ipfs::register_in_registry(&mut registry);
+        #[cfg(feature = "services-ipmfs")]
+        Ipmfs::register_in_registry(&mut registry);
+        #[cfg(feature = "services-icloud")]
+        Icloud::register_in_registry(&mut registry);
+        #[cfg(feature = "services-memcached")]
+        Memcached::register_in_registry(&mut registry);
+        #[cfg(feature = "services-memory")]
+        Memory::register_in_registry(&mut registry);
+        #[cfg(feature = "services-mini-moka")]
+        MiniMoka::register_in_registry(&mut registry);
+        #[cfg(feature = "services-moka")]
+        Moka::register_in_registry(&mut registry);
+        #[cfg(feature = "services-monoiofs")]
+        Monoiofs::register_in_registry(&mut registry);
+        #[cfg(feature = "services-mysql")]
+        Mysql::register_in_registry(&mut registry);
+        #[cfg(feature = "services-obs")]
+        Obs::register_in_registry(&mut registry);
+        #[cfg(feature = "services-onedrive")]
+        Onedrive::register_in_registry(&mut registry);
+        #[cfg(feature = "services-postgresql")]
+        Postgresql::register_in_registry(&mut registry);
+        #[cfg(feature = "services-gdrive")]
+        Gdrive::register_in_registry(&mut registry);
+        #[cfg(feature = "services-oss")]
+        Oss::register_in_registry(&mut registry);
+        #[cfg(feature = "services-persy")]
+        Persy::register_in_registry(&mut registry);
+        #[cfg(feature = "services-redis")]
+        Redis::register_in_registry(&mut registry);
+        #[cfg(feature = "services-rocksdb")]
+        Rocksdb::register_in_registry(&mut registry);
+        #[cfg(feature = "services-s3")]
+        S3::register_in_registry(&mut registry);
+        #[cfg(feature = "services-seafile")]
+        Seafile::register_in_registry(&mut registry);
+        #[cfg(feature = "services-upyun")]
+        Upyun::register_in_registry(&mut registry);
+        #[cfg(feature = "services-yandex-disk")]
+        YandexDisk::register_in_registry(&mut registry);
+        #[cfg(feature = "services-pcloud")]
+        Pcloud::register_in_registry(&mut registry);
+        #[cfg(feature = "services-sftp")]
+        Sftp::register_in_registry(&mut registry);
+        #[cfg(feature = "services-sled")]
+        Sled::register_in_registry(&mut registry);
+        #[cfg(feature = "services-sqlite")]
+        Sqlite::register_in_registry(&mut registry);
+        #[cfg(feature = "services-swift")]
+        Swift::register_in_registry(&mut registry);
+        #[cfg(feature = "services-tikv")]
+        Tikv::register_in_registry(&mut registry);
+        #[cfg(feature = "services-vercel-artifacts")]
+        VercelArtifacts::register_in_registry(&mut registry);
+        #[cfg(feature = "services-vercel-blob")]
+        VercelBlob::register_in_registry(&mut registry);
+        #[cfg(feature = "services-webdav")]
+        Webdav::register_in_registry(&mut registry);
+        #[cfg(feature = "services-webhdfs")]
+        Webhdfs::register_in_registry(&mut registry);
+        #[cfg(feature = "services-redb")]
+        Redb::register_in_registry(&mut registry);
+        #[cfg(feature = "services-mongodb")]
+        Mongodb::register_in_registry(&mut registry);
+        #[cfg(feature = "services-hdfs-native")]
+        HdfsNative::register_in_registry(&mut registry);
+        #[cfg(feature = "services-surrealdb")]
+        Surrealdb::register_in_registry(&mut registry);
+        #[cfg(feature = "services-lakefs")]
+        Lakefs::register_in_registry(&mut registry);
+        #[cfg(feature = "services-nebula-graph")]
+        NebulaGraph::register_in_registry(&mut registry);
+
+        registry
+    }
+
+    /// TODO: document this
+    pub fn register<T: Into<String>>(&mut self, scheme: T, factory: OperatorFactory) {
+        self.registry.insert(scheme.into(), factory);
+    }
+
+    /// TODO: document this
+    pub fn parse(
+        &self,
+        uri: &str,
+        options: impl IntoIterator<Item = (String, String)>,
+    ) -> Result<Operator> {
+        let parsed_uri = uri.parse::<Uri>().map_err(|err| {
+            Error::new(ErrorKind::ConfigInvalid, "uri is invalid")
+                .with_context("uri", uri)
+                .set_source(err)
+        })?;
+
+        let scheme = parsed_uri.scheme_str().ok_or_else(|| {
+            Error::new(ErrorKind::ConfigInvalid, "uri is missing scheme").with_context("uri", uri)
+        })?;
+
+        let factory = self.registry.get(scheme).ok_or_else(|| {
+            Error::new(
+                ErrorKind::ConfigInvalid,
+                "could not find any operator factory for the given scheme",
+            )
+            .with_context("uri", uri)
+            .with_context("scheme", scheme)
+        })?;
+
+        let options = options.into_iter().collect();
+
+        factory(&parsed_uri, options)
+    }
+}