feat: provide API for using your own object store (#730)

Closes #643 

I'm still not convinced by my decision to use `Format` as the base for
all this IO, but at least it makes clear what format you're expecting to
be reading/writing stuff in so 🤷🏼? No immediate plans to refactor ...

In support of stac-utils/rustac-py#117

Author: gadomski

.github/workflows/ci.yml

@@ -8,7 +8,7 @@ on:
 
 concurrency:
   group: ${{ github.workflow }}-${{ github.ref }}
-  cancel-in-progress: true
+  cancel-in-progress: false
 
 env:
   CARGO_TERM_COLOR: always

crates/core/CHANGELOG.md

@@ -12,6 +12,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 - `Clone` for `Container` ([#666](https://github.com/stac-utils/rustac/pull/666))
 - `Serialize` for `Container` ([#667](https://github.com/stac-utils/rustac/pull/667))
 - More permissive datetime interval parsing ([#715](https://github.com/stac-utils/rustac/pull/715))
+- `Format` methods for providing your own object store ([#730](https://github.com/stac-utils/rustac/pull/730))
 
 ### Changed
 

crates/core/src/format.rs

@@ -163,15 +163,14 @@ impl Format {
         let href = href.into();
         match href.clone().realize() {
             RealizedHref::Url(url) => {
-                use object_store::ObjectStore;
-
                 let (object_store, path) = parse_url_opts(&url, options)?;
-                tracing::debug!("getting {self} from {url} with object store");
-                let get_result = object_store.get(&path).await.map_err(|err| Error::Get {
-                    href,
-                    message: err.to_string(),
-                })?;
-                let mut value: T = self.from_bytes(get_result.bytes().await?)?;
+                let mut value: T =
+                    self.get_store(object_store.into(), path)
+                        .await
+                        .map_err(|err| Error::Get {
+                            href,
+                            message: err.to_string(),
+                        })?;
                 *value.self_href_mut() = Some(Href::Url(url));
                 Ok(value)
             }
@@ -188,6 +187,23 @@ impl Format {
         }
     }
 
+    /// Gets a STAC value from an object store.
+    #[cfg(feature = "object-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 + FromJson + FromNdjson + FromGeoparquet,
+    {
+        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
@@ -257,17 +273,31 @@ impl Format {
     {
         let href = href.to_string();
         if let Ok(url) = url::Url::parse(&href) {
-            use object_store::ObjectStore;
-
             let (object_store, path) = parse_url_opts(&url, options)?;
-            let bytes = self.into_vec(value)?;
-            let put_result = object_store.put(&path, bytes.into()).await?;
-            Ok(Some(put_result))
+            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 = "object-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: ToJson + ToNdjson + IntoGeoparquet,
+    {
+        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)
@@ -384,6 +414,7 @@ impl FromStr for Format {
 #[cfg(test)]
 mod tests {
     use super::Format;
+    use crate::Item;
     use crate::geoparquet::Compression;
 
     #[test]
@@ -425,4 +456,33 @@ mod tests {
             Format::infer_from_href("out.parquet").unwrap()
         );
     }
+
+    #[tokio::test]
+    #[cfg(feature = "object-store")]
+    async fn prefix_store_read() {
+        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 = "object-store")]
+    async fn store_write() {
+        use object_store::ObjectStore;
+        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();
+    }
 }