feat: storage abstraction (#749)

## Closes

- #731

cc @kylebarron

Author: gadomski

crates/cli/src/lib.rs

@@ -4,7 +4,7 @@
 
 use anyhow::{Error, Result, anyhow};
 use clap::{Parser, Subcommand};
-use stac::{Collection, Item, Links, Migrate, geoparquet::Compression};
+use stac::{Collection, Href, Item, Links, Migrate, geoparquet::Compression};
 use stac_api::{GetItems, GetSearch, Search};
 use stac_io::{Format, Validate};
 use stac_server::Backend;
@@ -459,7 +459,8 @@ impl Rustac {
         let href = href.and_then(|s| if s == "-" { None } else { Some(s) });
         let format = self.input_format(href);
         if let Some(href) = href {
-            let value: stac::Value = format.get_opts(href, self.opts()).await?;
+            let (store, path) = stac_io::parse_href_opts(Href::from(href), self.opts())?;
+            let value: stac::Value = store.get_format(path, format).await?;
             Ok(value)
         } else {
             let mut buf = Vec::new();
@@ -473,10 +474,10 @@ impl Rustac {
         let href = href.and_then(|s| if s == "-" { None } else { Some(s) });
         let format = self.output_format(href);
         if let Some(href) = href {
-            let opts = self.opts();
+            let (store, path) = stac_io::parse_href_opts(Href::from(href), self.opts())?;
             let _ = match value {
-                Value::Json(json) => format.put_opts(href, json, opts).await?,
-                Value::Stac(stac) => format.put_opts(href, stac, opts).await?,
+                Value::Json(json) => store.put_format(path, json, format).await?,
+                Value::Stac(stac) => store.put_format(path, stac, format).await?,
             };
             Ok(())
         } else {

crates/core/src/href.rs

@@ -178,7 +178,16 @@ impl TryFrom<Href> for Url {
     fn try_from(value: Href) -> Result<Self> {
         match value {
             Href::Url(url) => Ok(url),
-            Href::String(s) => s.parse().map_err(Error::from),
+            Href::String(mut s) => {
+                if !s.starts_with("/") {
+                    s = std::env::current_dir()?
+                        .join(s)
+                        .to_string_lossy()
+                        .into_owned();
+                }
+                let url = Url::parse(&format!("file://{s}"))?;
+                Ok(url)
+            }
         }
     }
 }
@@ -322,3 +331,17 @@ fn make_relative(href: &str, base: &str) -> String {
 
     relative
 }
+
+#[cfg(test)]
+mod tests {
+    use super::Href;
+    use url::Url;
+
+    #[test]
+    fn href_to_url() {
+        let href = Href::from("examples/simple-item.json");
+        let url: Url = href.try_into().unwrap();
+        assert_eq!(url.scheme(), "file");
+        assert!(url.path().ends_with("examples/simple-item.json"));
+    }
+}

crates/io/src/error.rs

@@ -83,6 +83,10 @@ pub enum Error {
     #[cfg(feature = "validate")]
     #[error(transparent)]
     JsonschemaValidation(#[from] jsonschema::ValidationError<'static>),
+
+    /// [url::ParseError]
+    #[error(transparent)]
+    UrlParse(#[from] url::ParseError),
 }
 
 #[cfg(feature = "validate")]

crates/io/src/format.rs

@@ -128,75 +128,6 @@ impl Format {
         Ok(value)
     }
 
-    /// Gets a STAC value from an object store with the provided options.
-    ///
-    /// # Examples
-    ///
-    /// ```no_run
-    /// use stac::Catalog;
-    /// use stac_io::Format;
-    ///
-    /// #[cfg(feature = "store-aws")]
-    /// {
-    /// # tokio_test::block_on(async {
-    ///     let catalog: Catalog = stac_io::get_opts("s3://nz-elevation/catalog.json",
-    ///         [("skip_signature", "true"), ("region", "ap-southeast-2")],
-    ///     ).await.unwrap();
-    /// # })
-    /// }
-    /// ```
-    #[cfg(feature = "store")]
-    pub async fn get_opts<T, I, K, V>(&self, href: impl Into<Href>, options: I) -> Result<T>
-    where
-        T: SelfHref + Readable,
-        I: IntoIterator<Item = (K, V)>,
-        K: AsRef<str>,
-        V: Into<String>,
-    {
-        let href = href.into();
-        match href.clone().into() {
-            RealizedHref::Url(url) => {
-                let (object_store, path) = parse_url_opts(&url, options)?;
-                let mut value: T =
-                    self.get_store(object_store.into(), path)
-                        .await
-                        .map_err(|err| Error::Get {
-                            href,
-                            message: err.to_string(),
-                        })?;
-                value.set_self_href(url);
-                Ok(value)
-            }
-            RealizedHref::PathBuf(path) => {
-                tracing::debug!(
-                    "getting {self} from {} with the standard library",
-                    path.display()
-                );
-                self.from_path(path).map_err(|err| Error::Get {
-                    href,
-                    message: err.to_string(),
-                })
-            }
-        }
-    }
-
-    /// Gets a STAC value from an object store.
-    #[cfg(feature = "store")]
-    pub async fn get_store<T>(
-        &self,
-        object_store: std::sync::Arc<dyn object_store::ObjectStore>,
-        path: impl Into<object_store::path::Path>,
-    ) -> Result<T>
-    where
-        T: SelfHref + Readable,
-    {
-        let path = path.into();
-        tracing::debug!("getting {self} from {path} with object store");
-        let get_result = object_store.get(&path).await?;
-        let value: T = self.from_bytes(get_result.bytes().await?)?;
-        Ok(value)
-    }
-
     /// Writes a STAC value to the provided path.
     ///
     /// # Examples
@@ -237,62 +168,6 @@ impl Format {
         Ok(value)
     }
 
-    /// Puts a STAC value to an object store with the provided options.
-    ///
-    /// # Examples
-    ///
-    /// ```no_run
-    /// use stac::Item;
-    /// use stac_io::Format;
-    ///
-    /// let item = Item::new("an-id");
-    /// #[cfg(feature = "store-aws")]
-    /// {
-    /// # tokio_test::block_on(async {
-    ///     Format::json().put_opts("s3://bucket/item.json", item, [("aws_access_key_id", "...")]).await.unwrap();
-    /// # })
-    /// }
-    /// ```
-    #[cfg(feature = "store")]
-    pub async fn put_opts<T, I, K, V>(
-        &self,
-        href: impl ToString,
-        value: T,
-        options: I,
-    ) -> Result<Option<object_store::PutResult>>
-    where
-        T: Writeable,
-        I: IntoIterator<Item = (K, V)>,
-        K: AsRef<str>,
-        V: Into<String>,
-    {
-        let href = href.to_string();
-        if let Ok(url) = url::Url::parse(&href) {
-            let (object_store, path) = parse_url_opts(&url, options)?;
-            self.put_store(object_store.into(), path, value)
-                .await
-                .map(Some)
-        } else {
-            self.write(href, value).map(|_| None)
-        }
-    }
-
-    /// Puts a STAC value into an object store.
-    #[cfg(feature = "store")]
-    pub async fn put_store<T>(
-        &self,
-        object_store: std::sync::Arc<dyn object_store::ObjectStore>,
-        path: impl Into<object_store::path::Path>,
-        value: T,
-    ) -> Result<object_store::PutResult>
-    where
-        T: Writeable,
-    {
-        let bytes = self.into_vec(value)?;
-        let put_result = object_store.put(&path.into(), bytes.into()).await?;
-        Ok(put_result)
-    }
-
     /// Returns the default JSON format (compact).
     pub fn json() -> Format {
         Format::Json(false)
@@ -317,33 +192,6 @@ impl Format {
     }
 }
 
-#[cfg(feature = "store")]
-fn parse_url_opts<I, K, V>(
-    url: &url::Url,
-    options: I,
-) -> Result<(Box<dyn object_store::ObjectStore>, object_store::path::Path)>
-where
-    I: IntoIterator<Item = (K, V)>,
-    K: AsRef<str>,
-    V: Into<String>,
-{
-    // It's technically inefficient to parse it twice, but we're doing this to
-    // then do IO so who cares.
-    #[cfg(feature = "store-aws")]
-    if let Ok((object_store::ObjectStoreScheme::AmazonS3, path)) =
-        object_store::ObjectStoreScheme::parse(url)
-    {
-        let mut builder = object_store::aws::AmazonS3Builder::from_env();
-        for (key, value) in options {
-            builder = builder.with_config(key.as_ref().parse()?, value);
-        }
-        return Ok((Box::new(builder.with_url(url.to_string()).build()?), path));
-    }
-
-    let result = object_store::parse_url_opts(url, options)?;
-    Ok(result)
-}
-
 impl Default for Format {
     fn default() -> Self {
         Self::Json(false)
@@ -461,35 +309,4 @@ mod tests {
             );
         }
     }
-
-    #[tokio::test]
-    #[cfg(feature = "store")]
-    async fn prefix_store_read() {
-        use stac::Item;
-        use std::sync::Arc;
-
-        let object_store =
-            object_store::local::LocalFileSystem::new_with_prefix("examples").unwrap();
-        let _: Item = Format::json()
-            .get_store(Arc::new(object_store), "simple-item.json")
-            .await
-            .unwrap();
-    }
-
-    #[tokio::test]
-    #[cfg(feature = "store")]
-    async fn store_write() {
-        use object_store::ObjectStore;
-        use stac::Item;
-        use std::sync::Arc;
-
-        let object_store = Arc::new(object_store::memory::InMemory::new());
-        let item = Item::new("an-id");
-        let _ = Format::json()
-            .put_store(object_store.clone(), "item.json", item)
-            .await
-            .unwrap();
-        let get_result = object_store.get(&"item.json".into()).await.unwrap();
-        let _: Item = serde_json::from_slice(&get_result.bytes().await.unwrap()).unwrap();
-    }
 }

crates/io/src/lib.rs

@@ -15,10 +15,7 @@ mod write;
 #[cfg(feature = "geoparquet")]
 pub use geoparquet::{FromGeoparquetPath, IntoGeoparquetPath};
 #[cfg(feature = "store")]
-pub use store::{
-    get::{get, get_opts},
-    put::{put, put_opts},
-};
+pub use store::{StacStore, parse_href, parse_href_opts};
 #[cfg(feature = "validate")]
 pub use validate::{Validate, Validator};
 pub use {
@@ -134,18 +131,6 @@ mod tests {
         let _ = super::read::<ItemCollection>("data/extended-item.parquet").unwrap_err();
     }
 
-    #[tokio::test]
-    #[cfg(all(feature = "store", not(target_os = "windows")))]
-    async fn get() {
-        let path = format!(
-            "file://{}",
-            std::fs::canonicalize("examples/simple-item.json")
-                .unwrap()
-                .to_string_lossy()
-        );
-        let _: Item = super::get(path).await.unwrap();
-    }
-
     #[test]
     fn write() {
         let tempdir = TempDir::new().unwrap();
@@ -154,18 +139,4 @@ mod tests {
         let item: Item = super::read(tempdir.path().join("item.json")).unwrap();
         assert_eq!(item.id, "an-id");
     }
-
-    #[tokio::test]
-    #[cfg(feature = "store")]
-    async fn put() {
-        let tempdir = TempDir::new().unwrap();
-        let path = format!(
-            "file://{}",
-            tempdir.path().join("item.json").to_string_lossy()
-        );
-        let item = Item::new("an-id");
-        assert!(super::put(path, item).await.unwrap().is_some());
-        let item: Item = crate::read(tempdir.path().join("item.json")).unwrap();
-        assert_eq!(item.id, "an-id");
-    }
 }