refactor(core)!: move a bunch of io around

Author: gadomski

.gitignore

@@ -1,3 +1,4 @@
 /target
 Cargo.lock
 dist/
+pyrightconfig.json

cli/src/args/item.rs

@@ -3,6 +3,7 @@ use crate::{Result, Value};
 use stac::{item::Builder, Asset};
 use std::path::Path;
 use tokio::sync::mpsc::Sender;
+use url::Url;
 
 /// Arguments for the `item` subcommand.
 #[derive(clap::Args, Debug)]
@@ -33,14 +34,12 @@ pub(crate) struct Args {
 
 impl Run for Args {
     async fn run(self, _: Input, _: Option<Sender<Value>>) -> Result<Option<Value>> {
-        let (id, href): (Option<String>, Option<String>) = if stac::href_to_url(&self.id_or_href)
-            .is_none()
-            && !Path::new(&self.id_or_href).exists()
-        {
-            (Some(self.id_or_href), None)
-        } else {
-            (None, Some(self.id_or_href))
-        };
+        let (id, href): (Option<String>, Option<String>) =
+            if Url::parse(&self.id_or_href).is_err() && !Path::new(&self.id_or_href).exists() {
+                (Some(self.id_or_href), None)
+            } else {
+                (None, Some(self.id_or_href))
+            };
         let id = id
             .or_else(|| {
                 Path::new(href.as_ref().expect("if id is none, href should exist"))

cli/src/args/mod.rs

@@ -12,9 +12,8 @@ mod validate;
 
 use crate::{input::Input, options::KeyValue, output::Output, Result, Value};
 use clap::Parser;
-use stac::Format;
-use tokio::sync::mpsc::Sender;
-use tokio::task::JoinHandle;
+use stac::io::Format;
+use tokio::{sync::mpsc::Sender, task::JoinHandle};
 use tracing::metadata::Level;
 
 const BUFFER: usize = 100;

cli/src/args/search.rs

@@ -185,22 +185,21 @@ impl Run for Args {
         } else {
             Some(self.collections)
         };
-        #[cfg(feature = "duckdb")]
+        #[cfg(all(feature = "duckdb", feature = "geoparquet"))]
         {
-            if self
-                .duckdb
-                .unwrap_or_else(|| stac::geoparquet::has_extension(&self.href))
-            {
+            if self.duckdb.unwrap_or_else(|| {
+                matches!(
+                    stac::io::Format::infer_from_href(&self.href),
+                    Some(stac::io::Format::Geoparquet(_))
+                )
+            }) {
                 search_geoparquet(self.href, search, stream, self.max_items).await
             } else {
                 search_api(self.href, search, stream, self.max_items).await
             }
         }
         #[cfg(not(feature = "duckdb"))]
         {
-            if stac::geoparquet::has_extension(&self.href) {
-                tracing::warn!("'{}' has a geoparquet extension, but the `duckdb` feature is not enabled — attempting to search with an ApiClient", self.href);
-            }
             search_api(self.href, search, stream, self.max_items).await
         }
     }

cli/src/input.rs

@@ -1,10 +1,13 @@
+use std::io::Read;
+
 use crate::{options::Options, Error, Result};
-use stac::{io::Config, Format, Value};
+use stac::{io::Format, Value};
 
 /// The input to a CLI run.
 #[derive(Debug, Default)]
 pub(crate) struct Input {
-    config: Config,
+    format: Option<Format>,
+    options: Options,
     href: Option<String>,
 }
 
@@ -18,25 +21,34 @@ impl Input {
         let href = href
             .into()
             .and_then(|href| if href == "-" { None } else { Some(href) });
-        let config = Config::new().format(format).options(options.into());
-        Input { config, href }
+        Input {
+            format: format.into(),
+            href,
+            options: options.into(),
+        }
     }
 
     /// Creates a new input with the given href.
     pub(crate) fn with_href(&self, href: impl Into<Option<String>>) -> Input {
         Input {
-            config: self.config.clone(),
+            format: self.format,
             href: href.into(),
+            options: self.options.clone(),
         }
     }
 
     /// Gets a STAC value from the input.
     pub(crate) async fn get(&self) -> Result<Value> {
-        if let Some(href) = self.href.as_ref() {
-            self.config.get(href.clone()).await.map_err(Error::from)
+        if let Some(href) = self.href.as_deref() {
+            stac::io::get_format_opts(href, self.format, self.options.iter())
+                .await
+                .map_err(Error::from)
         } else {
-            self.config
-                .from_reader(std::io::stdin())
+            let mut buf = Vec::new();
+            let _ = std::io::stdin().read_to_end(&mut buf);
+            self.format
+                .unwrap_or_default()
+                .from_bytes(buf.into())
                 .map_err(Error::from)
         }
     }

cli/src/options.rs

@@ -11,6 +11,12 @@ pub(crate) struct KeyValue {
     value: String,
 }
 
+impl Options {
+    pub(crate) fn iter(&self) -> impl Iterator<Item = (&str, &str)> {
+        self.0.iter().map(|kv| (kv.key.as_str(), kv.value.as_str()))
+    }
+}
+
 impl FromStr for KeyValue {
     type Err = Infallible;
     fn from_str(s: &str) -> Result<Self, Infallible> {

cli/src/output.rs

@@ -1,9 +1,9 @@
 //! Structures for writing output data.
 
 use crate::{options::Options, value::Value, Error, Result};
-use object_store::{local::LocalFileSystem, ObjectStore, PutResult};
-use stac::{io::Config, Format};
-use std::{io::IsTerminal, path::Path, pin::Pin};
+use object_store::PutResult;
+use stac::io::{Format, IntoFormattedBytes};
+use std::{path::Path, pin::Pin};
 use tokio::{
     fs::File,
     io::{AsyncWrite, AsyncWriteExt},
@@ -15,7 +15,7 @@ pub(crate) struct Output {
     pub(crate) format: Format,
     href: Option<String>,
     stream: Pin<Box<dyn AsyncWrite + Send>>,
-    config: Config,
+    options: Options,
 }
 
 impl Output {
@@ -26,10 +26,9 @@ impl Output {
         options: impl Into<Options>,
         create_parent_directories: bool,
     ) -> Result<Output> {
-        let mut format = format
+        let format = format
             .or_else(|| href.as_deref().and_then(Format::infer_from_href))
             .unwrap_or_default();
-        let config = Config::new().format(Some(format)).options(options.into());
         let stream = if let Some(href) = href.as_deref() {
             if let Ok(url) = Url::parse(href) {
                 if url.scheme() == "file" {
@@ -39,53 +38,39 @@ impl Output {
                     )
                     .await?
                 } else {
-                    Box::pin(config.buf_writer(&url)?)
+                    unimplemented!("streaming to object stores is not supported");
+                    // FIXME turn this into an actual error
                 }
             } else {
                 create_file_stream(href, create_parent_directories).await?
             }
         } else {
-            if std::io::stdout().is_terminal() {
-                format = format.pretty();
-            }
             Box::pin(tokio::io::stdout())
         };
         Ok(Output {
             href,
             format,
             stream,
-            config,
+            options: options.into(),
         })
     }
 
     /// Streams a value to the output
     pub(crate) async fn stream(&mut self, value: Value) -> Result<()> {
-        let bytes = value.into_ndjson()?;
+        let bytes = value.into_formatted_bytes(Format::NdJson)?;
         self.stream.write_all(&bytes).await?;
         self.stream.flush().await?;
         Ok(())
     }
 
     /// Puts a value to the output.
     pub(crate) async fn put(&mut self, value: Value) -> Result<Option<PutResult>> {
-        let bytes = match value {
-            Value::Json(value) => self.format.json_to_vec(value)?,
-            Value::Stac(value) => self.format.value_to_vec(value)?,
-        };
         if let Some(href) = self.href.as_deref() {
-            let (object_store, path): (Box<dyn ObjectStore>, _) = match Url::parse(href) {
-                Ok(url) => self.config.object_store(&url)?,
-                Err(_) => {
-                    let path = object_store::path::Path::from_filesystem_path(href)?;
-                    (Box::new(LocalFileSystem::new()), path)
-                }
-            };
-            object_store
-                .put(&path, bytes.into())
+            stac::io::put_format_opts(href, value, self.format, self.options.iter())
                 .await
-                .map(Some)
                 .map_err(Error::from)
         } else {
+            let bytes = value.into_formatted_bytes(self.format)?;
             self.stream.write_all(&bytes).await?;
             self.stream.flush().await?;
             Ok(None)

cli/src/value.rs

@@ -1,6 +1,6 @@
 use crate::{Error, Result};
 use serde::Serialize;
-use stac::Format;
+use stac::io::{Format, IntoFormattedBytes};
 
 /// An output value, which can either be a [serde_json::Value] or a [stac::Value].
 #[derive(Debug, Serialize)]
@@ -13,15 +13,6 @@ pub enum Value {
     Json(serde_json::Value),
 }
 
-impl Value {
-    pub(crate) fn into_ndjson(self) -> Result<Vec<u8>> {
-        match self {
-            Value::Json(value) => Format::NdJson.json_to_vec(value).map_err(Error::from),
-            Value::Stac(value) => Format::NdJson.value_to_vec(value).map_err(Error::from),
-        }
-    }
-}
-
 impl From<stac::Value> for Value {
     fn from(value: stac::Value) -> Self {
         Value::Stac(value)
@@ -49,3 +40,12 @@ impl TryFrom<Value> for stac::Value {
         }
     }
 }
+
+impl IntoFormattedBytes for Value {
+    fn into_formatted_bytes(self, format: Format) -> stac::Result<Vec<u8>> {
+        match self {
+            Self::Json(value) => value.into_formatted_bytes(format),
+            Self::Stac(value) => value.into_formatted_bytes(format),
+        }
+    }
+}

core/CHANGELOG.md

@@ -9,18 +9,14 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 ### Added
 
 - Deref `ItemCollection` ([#363](https://github.com/stac-utils/stac-rs/pull/363))
-- `object_store` and `ndjson` ([#369](https://github.com/stac-utils/stac-rs/pull/369))
-- `Format` ([#372](https://github.com/stac-utils/stac-rs/pull/371))
+- `io::Format` ([#372](https://github.com/stac-utils/stac-rs/pull/371))
 - Read unknown versions ([#378](https://github.com/stac-utils/stac-rs/pull/378))
+- `io::IntoFormattedBytes` ([#386](https://github.com/stac-utils/stac-rs/pull/386))
 
 ### Changed
 
 - Update **geoarrow** to v0.3.0 ([#367](https://github.com/stac-utils/stac-rs/pull/367))
 
-### Removed
-
-- `io` module ([#369](https://github.com/stac-utils/stac-rs/pull/369))
-
 ## [0.9.0] - 2024-09-05
 
 ### Added

core/Cargo.toml

@@ -21,7 +21,7 @@ geoarrow = [
     "dep:arrow-schema",
     "dep:geo-types",
 ]
-geoparquet = ["geoarrow", "geoarrow/parquet", "dep:parquet", "dep:bytes"]
+geoparquet = ["geoarrow", "geoarrow/parquet", "dep:parquet"]
 geoparquet-compression = [
     "geoparquet",
     "geoarrow/parquet_compression",
@@ -31,7 +31,7 @@ geoparquet-compression = [
     "parquet/lz4",
     "parquet/zstd",
 ]
-object-store = ["dep:object_store", "dep:send-future"]
+object-store = ["dep:object_store"]
 object-store-aws = ["object-store", "object_store/aws"]
 object-store-azure = ["object-store", "object_store/azure"]
 object-store-gcp = ["object-store", "object_store/gcp"]
@@ -49,7 +49,7 @@ arrow-array = { version = "52", optional = true }
 arrow-cast = { version = "52", optional = true }
 arrow-json = { version = "52", optional = true }
 arrow-schema = { version = "52", optional = true }
-bytes = { version = "1", optional = true }
+bytes = "1"
 chrono = { version = "0.4", features = ["serde"] }
 gdal = { version = "0.17", optional = true }
 gdal-sys = { version = "0.10", optional = true }
@@ -62,7 +62,6 @@ mime = "0.3"
 object_store = { version = "0.11", optional = true }
 parquet = { version = "52", default-features = false, optional = true }
 reqwest = { version = "0.12", optional = true, features = ["json", "blocking"] }
-send-future = { version = "0.1", optional = true }                               # https://github.com/rust-lang/rust/issues/96865
 serde = { version = "1", features = ["derive"] }
 serde_json = { version = "1", features = ["preserve_order"] }
 thiserror = "1"
@@ -72,7 +71,6 @@ url = "2"
 assert-json-diff = "2"
 bytes = "1"
 rstest = "0.22"
-tokio = { version = "1", features = ["macros"] }
 tokio-test = "0.4"
 
 [package.metadata.docs.rs]