Repository: stac-utils/stac-rs (language: python)

Issue:
Use object_store in cli
Should be easy b/c https://docs.rs/object_store/latest/object_store/buffered/struct.BufWriter.html ... we might want to get smarter about gets, but we'll see.


Discussion / hints:

Fix patch:
diff --git a/cli/CHANGELOG.md b/cli/CHANGELOG.md
index f6b86ac6..cf14d659 100644
--- a/cli/CHANGELOG.md
+++ b/cli/CHANGELOG.md
@@ -10,6 +10,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 ### Added
 
 - Outfile and stream arguments to `items` ([#363](https://github.com/stac-utils/stac-rs/pull/363))
+- `object_store` ([#371](https://github.com/stac-utils/stac-rs/pull/371))
 
 ### Fixed
 
diff --git a/cli/Cargo.toml b/cli/Cargo.toml
index 85d98854..6468786f 100644
--- a/cli/Cargo.toml
+++ b/cli/Cargo.toml
@@ -15,37 +15,34 @@ rust-version = "1.75"
 default = ["gdal", "geoparquet", "pgstac"]
 duckdb = ["dep:stac-duckdb", "dep:duckdb"]
 gdal = ["stac/gdal"]
-geoparquet = [
-    "dep:bytes",
-    "dep:geoarrow",
-    "stac/geoparquet-compression",
-    "parquet",
-]
+geoparquet = ["dep:geoarrow", "stac/geoparquet-compression", "parquet"]
 pgstac = ["stac-server/pgstac"]
 python = ["dep:pyo3", "pgstac", "geoparquet"]
 
 [dependencies]
 axum = "0.7"
-bytes = { version = "1", optional = true }
+bytes = { version = "1" }
 clap = { version = "4", features = ["derive"] }
-duckdb = { version = "1", optional = true }                                  # We have this dependency only to allow us to bundle it
+duckdb = { version = "1", optional = true }                                      # We have this dependency only to allow us to bundle it
 geoarrow = { version = "0.3", optional = true }
+object_store = { version = "0.11", features = ["aws", "gcp", "azure", "http"] }
 parquet = { version = "52", optional = true }
 pyo3 = { version = "0.22", optional = true }
 reqwest = "0.12"
 serde = "1"
 serde_json = "1"
-stac = { version = "0.9.0", path = "../core" }
+stac = { version = "0.9.0", path = "../core", features = ["object_store"] }
 stac-api = { version = "0.5.0", path = "../api" }
 stac-async = { version = "0.5.3", path = "../async" }
 stac-duckdb = { version = "0.0.1", path = "../duckdb", optional = true }
 stac-server = { version = "0.2.0", path = "../server", features = ["axum"] }
 stac-validate = { version = "0.2.2", path = "../validate" }
 thiserror = "1"
-tokio = { version = "1.23", features = ["macros", "rt-multi-thread"] }
+tokio = { version = "1.23", features = ["macros", "io-std", "rt-multi-thread"] }
 tokio-stream = "0.1"
 tracing = "0.1"
 tracing-subscriber = "0.3"
+url = "2"
 
 [dev-dependencies]
 assert_cmd = "2"
diff --git a/cli/src/args/item.rs b/cli/src/args/item.rs
index 8599107f..5ce8770c 100644
--- a/cli/src/args/item.rs
+++ b/cli/src/args/item.rs
@@ -6,7 +6,7 @@ use tokio::sync::mpsc::Sender;
 
 /// Arguments for the `item` subcommand.
 #[derive(clap::Args, Debug)]
-pub struct Args {
+pub(crate) struct Args {
     /// The item id or asset href
     pub(crate) id_or_href: String,
 
@@ -32,7 +32,7 @@ pub struct Args {
 }
 
 impl Run for Args {
-    async fn run(self, _: Input, _: Sender<Value>) -> Result<Option<Value>> {
+    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()
diff --git a/cli/src/args/items.rs b/cli/src/args/items.rs
index 2dc8d7ee..086cfec5 100644
--- a/cli/src/args/items.rs
+++ b/cli/src/args/items.rs
@@ -4,7 +4,7 @@ use tokio::{sync::mpsc::Sender, task::JoinSet};
 
 /// Arguments for the `items` subcommand.
 #[derive(clap::Args, Debug)]
-pub struct Args {
+pub(crate) struct Args {
     /// The asset hrefs
     hrefs: Vec<String>,
 
@@ -12,10 +12,6 @@ pub struct Args {
     #[arg(long)]
     outfile: Option<String>,
 
-    /// Stream the items as they are created as ndjson
-    #[arg(short, long)]
-    stream: bool,
-
     /// The assets' key
     #[arg(short, long, default_value = "data")]
     key: String,
@@ -35,12 +31,12 @@ pub struct Args {
 }
 
 impl Run for Args {
-    async fn run(self, input: Input, sender: Sender<Value>) -> Result<Option<Value>> {
+    async fn run(self, input: Input, stream: Option<Sender<Value>>) -> Result<Option<Value>> {
         let mut join_set = JoinSet::new();
         let mut items = Vec::with_capacity(self.hrefs.len());
         for href in self.hrefs {
-            let input = input.clone();
-            let sender = sender.clone();
+            let input = input.with_href(&href)?;
+            let sender = stream.clone();
             let args = ItemArgs {
                 id_or_href: href,
                 outfile: None,
@@ -54,14 +50,14 @@ impl Run for Args {
         }
         while let Some(result) = join_set.join_next().await {
             if let Some(Value::Stac(stac::Value::Item(item))) = result?? {
-                if self.stream {
-                    sender.send(stac::Value::Item(item).into()).await?;
+                if let Some(ref stream) = stream {
+                    stream.send(stac::Value::Item(item).into()).await?;
                 } else {
                     items.push(item);
                 }
             }
         }
-        if self.stream {
+        if stream.is_some() {
             Ok(None)
         } else {
             Ok(Some(stac::Value::ItemCollection(items.into()).into()))
diff --git a/cli/src/args/migrate.rs b/cli/src/args/migrate.rs
index fdcc60e1..2e03b0f6 100644
--- a/cli/src/args/migrate.rs
+++ b/cli/src/args/migrate.rs
@@ -5,7 +5,7 @@ use tokio::sync::mpsc::Sender;
 
 /// Arguments for the `migrate` subcommand.
 #[derive(clap::Args, Debug)]
-pub struct Args {
+pub(crate) struct Args {
     /// The input file, if not provided or `-` the input will be read from standard input
     infile: Option<String>,
 
@@ -18,8 +18,8 @@ pub struct Args {
 }
 
 impl Run for Args {
-    async fn run(self, input: Input, _: Sender<Value>) -> Result<Option<Value>> {
-        let value = input.read(self.infile)?;
+    async fn run(self, input: Input, _: Option<Sender<Value>>) -> Result<Option<Value>> {
+        let value = input.get().await?;
         value
             .migrate(self.version)
             .map(Value::from)
@@ -27,6 +27,10 @@ impl Run for Args {
             .map_err(Error::from)
     }
 
+    fn take_infile(&mut self) -> Option<String> {
+        self.infile.take()
+    }
+
     fn take_outfile(&mut self) -> Option<String> {
         self.outfile.take()
     }
diff --git a/cli/src/args/mod.rs b/cli/src/args/mod.rs
index f3f8e96c..bc627e6c 100644
--- a/cli/src/args/mod.rs
+++ b/cli/src/args/mod.rs
@@ -10,15 +10,11 @@ mod serve;
 mod translate;
 mod validate;
 
-use crate::{Format, Result, Value};
+use crate::{config::Entry, input::Input, output::Output, Result, Value};
 use clap::Parser;
-use std::fs::File;
-use std::{
-    io::{BufWriter, Write},
-    path::Path,
-};
 use tokio::sync::mpsc::Sender;
 use tokio::task::JoinHandle;
+use tracing::info;
 use tracing::metadata::Level;
 
 const BUFFER: usize = 100;
@@ -29,20 +25,27 @@ const BUFFER: usize = 100;
 pub struct Args {
     /// The input format, if not provided will be inferred from the input file's extension, falling back to json
     #[arg(short, long, global = true)]
-    input_format: Option<Format>,
+    input_format: Option<stac::Format>,
+
+    /// key=value pairs to use for the input object store
+    #[arg(short = 'k', long)]
+    input_config: Vec<Entry>,
 
     /// The output format, if not provided will be inferred from the output file's extension, falling back to json
     #[arg(short, long, global = true)]
-    output_format: Option<Format>,
+    output_format: Option<crate::output::Format>,
+
+    /// key=value pairs to use for the output object store
+    #[arg(short = 'c', long)]
+    output_config: Vec<Entry>,
 
-    /// If output is being written to a file, create any directories in that file's path
-    #[arg(long, global = true)]
-    create_directories: bool,
+    /// If the output is a local file, create its parent directories before creating the file
+    #[arg(long, default_value_t = true)]
+    create_parent_directories: bool,
 
-    /// The parquet compression to use when writing stac-geoparquet
-    #[arg(long, global = true, help = parquet_compression_help(), long_help = parquet_compression_long_help())]
-    #[cfg(feature = "geoparquet")]
-    parquet_compression: Option<parquet::basic::Compression>,
+    /// Stream the items to output as ndjson, default behavior is to return them all at the end of the operation
+    #[arg(short, long)]
+    stream: bool,
 
     #[arg(
         long,
@@ -70,10 +73,10 @@ pub struct Args {
     subcommand: Subcommand,
 }
 
-/// A sucommand.
+/// A subcommand.
 #[derive(Debug, clap::Subcommand)]
 #[allow(clippy::large_enum_variant)]
-pub enum Subcommand {
+enum Subcommand {
     /// Create a STAC Item from an id or the href to an asset
     Item(item::Args),
 
@@ -99,13 +102,12 @@ pub enum Subcommand {
 #[derive(Copy, Clone, Debug, Default)]
 struct ErrorLevel;
 
-#[derive(Clone, Debug)]
-struct Input {
-    format: Option<Format>,
-}
-
 trait Run {
-    async fn run(self, input: Input, sender: Sender<Value>) -> Result<Option<Value>>;
+    async fn run(self, input: Input, stream: Option<Sender<Value>>) -> Result<Option<Value>>;
+
+    fn take_infile(&mut self) -> Option<String> {
+        None
+    }
 
     fn take_outfile(&mut self) -> Option<String> {
         None
@@ -123,63 +125,80 @@ impl Args {
     }
 
     /// Runs whatever these arguments say that we should run.
-    #[cfg_attr(not(feature = "geoparquet"), allow(unused_mut))]
-    pub async fn run(mut self, output: impl Write + Send + 'static) -> Result<()> {
-        let (mut writer, mut format): (Box<dyn Write + Send>, Format) =
-            if let Some(outfile) = self.subcommand.take_outfile() {
-                if self.create_directories && stac::href_to_url(&outfile).is_none() {
-                    let path = Path::new(&outfile);
-                    if let Some(parent) = path.parent() {
-                        std::fs::create_dir_all(parent)?;
-                    }
+    pub async fn run(mut self) -> Result<()> {
+        let input = Input::new(
+            self.subcommand.take_infile(),
+            self.input_format,
+            self.input_config,
+        )?;
+        let mut output = Output::new(
+            self.subcommand.take_outfile(),
+            self.output_format.or_else(|| {
+                if self.stream {
+                    Some(crate::output::Format::NdJson)
+                } else {
+                    None
                 }
-                let format = self
-                    .output_format
-                    .or_else(|| Format::infer(&outfile))
-                    .unwrap_or(Format::CompactJson);
-                (Box::new(BufWriter::new(File::create(outfile)?)), format)
-            } else {
-                let format = self.output_format.unwrap_or(Format::PrettyJson);
-                (Box::new(output), format)
-            };
-        #[cfg(feature = "geoparquet")]
-        if let Format::Geoparquet(ref mut compression) = format {
-            *compression = self.parquet_compression;
-        }
-        let (sender, mut receiver) = tokio::sync::mpsc::channel(BUFFER);
-        let receiver: JoinHandle<Result<_>> = tokio::task::spawn(async move {
-            while let Some(value) = receiver.recv().await {
-                Format::Streaming.to_writer(&mut writer, value)?;
+            }),
+            self.output_config,
+            self.create_parent_directories,
+        )
+        .await?;
+        let value = if self.stream {
+            if output.format != crate::output::Format::NdJson {
+                tracing::warn!(
+                    "format was set to {}, but stream=true so re-setting to nd-json",
+                    output.format
+                );
             }
-            Ok((writer, format))
-        });
-        let value = self
-            .subcommand
-            .run(
-                Input {
-                    format: self.input_format,
-                },
-                sender,
-            )
-            .await?;
-        let (writer, format) = receiver.await??;
+            let (stream, mut receiver) = tokio::sync::mpsc::channel(BUFFER);
+            let streamer: JoinHandle<Result<_>> = tokio::task::spawn(async move {
+                while let Some(value) = receiver.recv().await {
+                    output.stream(value).await?;
+                }
+                Ok(output)
+            });
+            let value = self.subcommand.run(input, Some(stream)).await?;
+            output = streamer.await??;
+            value
+        } else {
+            self.subcommand.run(input, None).await?
+        };
         if let Some(value) = value {
-            format.to_writer(writer, value)?;
+            if let Some(put_result) = output.put(value).await? {
+                info!(
+                    "put result: etag={}, version={}",
+                    put_result.e_tag.as_deref().unwrap_or("<none>"),
+                    put_result.version.as_deref().unwrap_or("<none>")
+                );
+            }
         }
         Ok(())
     }
 }
 
 impl Run for Subcommand {
-    async fn run(self, input: Input, sender: Sender<Value>) -> Result<Option<Value>> {
+    async fn run(self, input: Input, stream: Option<Sender<Value>>) -> Result<Option<Value>> {
+        match self {
+            Subcommand::Item(args) => args.run(input, stream).await,
+            Subcommand::Items(args) => args.run(input, stream).await,
+            Subcommand::Migrate(args) => args.run(input, stream).await,
+            Subcommand::Search(args) => args.run(input, stream).await,
+            Subcommand::Serve(args) => args.run(input, stream).await,
+            Subcommand::Translate(args) => args.run(input, stream).await,
+            Subcommand::Validate(args) => args.run(input, stream).await,
+        }
+    }
+
+    fn take_infile(&mut self) -> Option<String> {
         match self {
-            Subcommand::Item(args) => args.run(input, sender).await,
-            Subcommand::Items(args) => args.run(input, sender).await,
-            Subcommand::Migrate(args) => args.run(input, sender).await,
-            Subcommand::Search(args) => args.run(input, sender).await,
-            Subcommand::Serve(args) => args.run(input, sender).await,
-            Subcommand::Translate(args) => args.run(input, sender).await,
-            Subcommand::Validate(args) => args.run(input, sender).await,
+            Subcommand::Item(args) => args.take_infile(),
+            Subcommand::Items(args) => args.take_infile(),
+            Subcommand::Migrate(args) => args.take_infile(),
+            Subcommand::Search(args) => args.take_infile(),
+            Subcommand::Serve(args) => args.take_infile(),
+            Subcommand::Translate(args) => args.take_infile(),
+            Subcommand::Validate(args) => args.take_infile(),
         }
     }
 
@@ -196,24 +215,6 @@ impl Run for Subcommand {
     }
 }
 
-impl Input {
-    fn read(self, infile: Option<String>) -> Result<stac::Value> {
-        let format = self
-            .format
-            .or_else(|| infile.as_deref().and_then(Format::infer))
-            .unwrap_or(Format::CompactJson);
-        if let Some(infile) =
-            infile.and_then(|infile| if infile == "-" { None } else { Some(infile) })
-        {
-            tracing::info!("reading '{}'", infile);
-            format.from_file(&infile)
-        } else {
-            tracing::info!("reading from standard input");
-            format.from_reader(std::io::stdin())
-        }
-    }
-}
-
 impl ErrorLevel {
     fn default() -> Option<Level> {
         Some(Level::ERROR)
@@ -257,23 +258,3 @@ fn level_value(level: Option<Level>) -> i8 {
         Some(Level::TRACE) => 4,
     }
 }
-
-#[cfg(feature = "geoparquet")]
-fn parquet_compression_help() -> &'static str {
-    "The parquet compression to use when writing stac-geoparquet [possible values: uncompressed, snappy, gzip(level), lzo, brotli(level), lz4, zstd(level), lz4_raw]"
-}
-
-#[cfg(feature = "geoparquet")]
-fn parquet_compression_long_help() -> &'static str {
-    "The parquet compression to use when writing stac-geoparquet
-
-Possible values:
-- uncompressed
-- snappy
-- gzip(level)
-- lzo
-- brotli(level)
-- lz4
-- zstd(level)
-- lz4_raw"
-}
diff --git a/cli/src/args/search.rs b/cli/src/args/search.rs
index 6b40a97f..845c9a04 100644
--- a/cli/src/args/search.rs
+++ b/cli/src/args/search.rs
@@ -9,7 +9,7 @@ use tokio_stream::StreamExt;
 use tracing::info;
 
 #[derive(Debug, clap::Args)]
-pub struct Args {
+pub(crate) struct Args {
     /// The href of the STAC API or the stac-geoparquet file to search
     #[cfg(feature = "duckdb")]
     href: String,
@@ -73,10 +73,6 @@ pub struct Args {
     #[arg(long)]
     query: Option<String>,
 
-    /// Stream the items to output as ndjson, default behavior is to return all the items as one item collection after the search is complete
-    #[arg(short, long)]
-    stream: bool,
-
     /// Whether to search with DuckDB instead of a STAC API client.
     ///
     /// ⚠ Experimental: this is a new feature whose behavior might change without warning.
@@ -164,7 +160,7 @@ async fn search_geoparquet(
 }
 
 impl Run for Args {
-    async fn run(self, _: Input, sender: Sender<Value>) -> Result<Option<Value>> {
+    async fn run(self, _: Input, stream: Option<Sender<Value>>) -> Result<Option<Value>> {
         let get_items = GetItems {
             bbox: self.bbox,
             datetime: self.datetime,
@@ -190,16 +186,15 @@ impl Run for Args {
         } else {
             Some(self.collections)
         };
-        let sender = if self.stream { Some(sender) } else { None };
         #[cfg(feature = "duckdb")]
         {
             if self
                 .duckdb
                 .unwrap_or_else(|| stac::geoparquet::has_extension(&self.href))
             {
-                search_geoparquet(self.href, search, sender, self.max_items).await
+                search_geoparquet(self.href, search, stream, self.max_items).await
             } else {
-                search_api(self.href, search, sender, self.max_items).await
+                search_api(self.href, search, stream, self.max_items).await
             }
         }
         #[cfg(not(feature = "duckdb"))]
@@ -207,7 +202,7 @@ impl Run for Args {
             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, sender, self.max_items).await
+            search_api(self.href, search, stream, self.max_items).await
         }
     }
 
diff --git a/cli/src/args/serve.rs b/cli/src/args/serve.rs
index 555ed54f..8e1ac91e 100644
--- a/cli/src/args/serve.rs
+++ b/cli/src/args/serve.rs
@@ -10,7 +10,7 @@ const DEFAULT_COLLECTION_ID: &str = "auto-generated-collection";
 
 /// Arguments for serving an API.
 #[derive(Debug, clap::Args)]
-pub struct Args {
+pub(crate) struct Args {
     /// Hrefs of collections, items, and item collections to load into the server on start
     ///
     /// If this is a single `-`, the data will be read from standard input.
@@ -44,7 +44,7 @@ struct Info {
 }
 
 impl Run for Args {
-    async fn run(self, input: Input, sender: Sender<Value>) -> Result<Option<Value>> {
+    async fn run(self, input: Input, _: Option<Sender<Value>>) -> Result<Option<Value>> {
         use stac::Value;
 
         let mut info = Info::default();
@@ -68,8 +68,7 @@ impl Run for Args {
         };
 
         // TODO maybe this is worth pulling out into something more general?
-        info!("loading items into the backend");
-        let mut join_set = JoinSet::new();
+        let mut join_set: JoinSet<Result<Value>> = JoinSet::new();
         let mut reading_from_stdin = false;
         for href in self.hrefs {
             if href == "-" {
@@ -79,8 +78,12 @@ impl Run for Args {
                     reading_from_stdin = true;
                 }
             }
-            let input = input.clone();
-            let _ = join_set.spawn(async move { input.read(Some(href)) });
+            let input = input.with_href(&href)?;
+            let _ = join_set.spawn(async move {
+                let mut value = input.get().await?;
+                value.set_href(href);
+                Ok(value)
+            });
         }
         let mut item_join_set = JoinSet::new();
         let mut collections = HashSet::new();
@@ -99,8 +102,8 @@ impl Run for Args {
                         collection.make_relative_links_absolute(href)?;
                         for link in collection.iter_item_links() {
                             let href = link.href.to_string();
-                            let input = input.clone();
-                            let _ = item_join_set.spawn(async move { input.read(Some(href)) });
+                            let input = input.with_href(&href)?;
+                            let _ = item_join_set.spawn(async move { input.get().await });
                         }
                     }
                     let _ = collections.insert(collection.id.clone());
@@ -151,18 +154,14 @@ impl Run for Args {
 
         info!("starting server");
         match backend {
-            Backend::Memory(backend) => start_server(backend, self.addr, sender, info)
-                .await
-                .and(Ok(None)),
+            Backend::Memory(backend) => start_server(backend, self.addr, info).await.and(Ok(None)),
             #[cfg(feature = "pgstac")]
-            Backend::Pgstac(backend) => start_server(backend, self.addr, sender, info)
-                .await
-                .and(Ok(None)),
+            Backend::Pgstac(backend) => start_server(backend, self.addr, info).await.and(Ok(None)),
         }
     }
 }
 
-async fn start_server<B>(backend: B, addr: String, sender: Sender<Value>, info: Info) -> Result<()>
+async fn start_server<B>(backend: B, addr: String, info: Info) -> Result<()>
 where
     B: stac_server::Backend,
 {
@@ -170,17 +169,12 @@ where
     let api = Api::new(backend, &root)?;
     let router = stac_server::routes::from_api(api);
     let listener = TcpListener::bind(addr).await?;
-    sender
-        .send(
-            format!(
-                "Serving a STAC API at {} using a {} backend{}",
-                root,
-                info.backend,
-                info.counts()
-            )
-            .into(),
-        )
-        .await?;
+    eprintln!(
+        "Serving a STAC API at {} using a {} backend{}",
+        root,
+        info.backend,
+        info.counts()
+    );
     axum::serve(listener, router).await.map_err(Error::from)
 }
 
diff --git a/cli/src/args/translate.rs b/cli/src/args/translate.rs
index f5c46f88..faf2c382 100644
--- a/cli/src/args/translate.rs
+++ b/cli/src/args/translate.rs
@@ -4,7 +4,7 @@ use tokio::sync::mpsc::Sender;
 
 /// Arguments for the `translate` subcommand.
 #[derive(clap::Args, Debug)]
-pub struct Args {
+pub(crate) struct Args {
     /// The input file, if not provided or `-` the input will be read from standard input
     infile: Option<String>,
 
@@ -13,8 +13,12 @@ pub struct Args {
 }
 
 impl Run for Args {
-    async fn run(self, input: Input, _: Sender<Value>) -> Result<Option<Value>> {
-        input.read(self.infile).map(|value| Some(value.into()))
+    async fn run(self, input: Input, _: Option<Sender<Value>>) -> Result<Option<Value>> {
+        input.get().await.map(|value| Some(value.into()))
+    }
+
+    fn take_infile(&mut self) -> Option<String> {
+        self.infile.take()
     }
 
     fn take_outfile(&mut self) -> Option<String> {
diff --git a/cli/src/args/validate.rs b/cli/src/args/validate.rs
index 2cf95fff..6b426232 100644
--- a/cli/src/args/validate.rs
+++ b/cli/src/args/validate.rs
@@ -5,7 +5,7 @@ use tokio::sync::mpsc::Sender;
 
 /// Arguments for the `validate` subcommand.
 #[derive(clap::Args, Debug)]
-pub struct Args {
+pub(crate) struct Args {
     /// The input file, if not provided or `-` the input will be read from standard input
     infile: Option<String>,
 
@@ -14,8 +14,8 @@ pub struct Args {
 }
 
 impl Run for Args {
-    async fn run(self, input: Input, sender: Sender<Value>) -> Result<Option<Value>> {
-        let value = input.read(self.infile)?;
+    async fn run(self, input: Input, _: Option<Sender<Value>>) -> Result<Option<Value>> {
+        let value = input.get().await?;
         let result = tokio::task::spawn_blocking(move || {
             if let Err(err) = value.validate() {
                 Err((err, value))
@@ -32,16 +32,19 @@ impl Run for Args {
                 stac::Value::ItemCollection(_) => "[item-collection] ".to_string(),
             };
             for error in errors {
-                let message = format!(
+                eprintln!(
                     "{}{} (instance path: '{}', schema path: '{}')",
                     message_base, error, error.instance_path, error.schema_path
                 );
-                sender.send(message.into()).await?;
             }
         }
         result.and(Ok(None)).map_err(|(err, _)| Error::from(err))
     }
 
+    fn take_infile(&mut self) -> Option<String> {
+        self.infile.take()
+    }
+
     fn take_outfile(&mut self) -> Option<String> {
         self.outfile.take()
     }
diff --git a/cli/src/config.rs b/cli/src/config.rs
new file mode 100644
index 00000000..aefb809a
--- /dev/null
+++ b/cli/src/config.rs
@@ -0,0 +1,44 @@
+use std::{convert::Infallible, str::FromStr};
+
+/// A collection of configuration entries.
+#[derive(Clone, Debug, Default)]
+pub(crate) struct Config(Vec<Entry>);
+
+/// `key=value`` pairs for object store configuration
+#[derive(Clone, Debug)]
+pub(crate) struct Entry {
+    key: String,
+    value: String,
+}
+
+impl Config {
+    /// Returns an iterator over this config's key value pairs.
+    pub(crate) fn iter(&self) -> impl Iterator<Item = (&str, &str)> {
+        self.0
+            .iter()
+            .map(|entry| (entry.key.as_str(), entry.value.as_str()))
+    }
+}
+
+impl FromStr for Entry {
+    type Err = Infallible;
+    fn from_str(s: &str) -> Result<Self, Infallible> {
+        if let Some((key, value)) = s.split_once('=') {
+            Ok(Entry {
+                key: key.to_string(),
+                value: value.to_string(),
+            })
+        } else {
+            Ok(Entry {
+                key: s.to_string(),
+                value: String::new(),
+            })
+        }
+    }
+}
+
+impl From<Vec<Entry>> for Config {
+    fn from(value: Vec<Entry>) -> Self {
+        Config(value)
+    }
+}
diff --git a/cli/src/error.rs b/cli/src/error.rs
index 1b13c534..f2f87467 100644
--- a/cli/src/error.rs
+++ b/cli/src/error.rs
@@ -9,6 +9,19 @@ pub enum Error {
     #[error(transparent)]
     Io(#[from] std::io::Error),
 
+    /// [object_store::Error]
+    #[error(transparent)]
+    ObjectStore(#[from] object_store::Error),
+
+    /// [object_store::path::Error]
+    #[error(transparent)]
+    ObjectStorePath(#[from] object_store::path::Error),
+
+    /// [parquet::errors::ParquetError]
+    #[error(transparent)]
+    #[cfg(feature = "geoparquet")]
+    Parquet(#[from] parquet::errors::ParquetError),
+
     /// [reqwest::Error]
     #[error(transparent)]
     Reqwest(#[from] reqwest::Error),
diff --git a/cli/src/format.rs b/cli/src/format.rs
deleted file mode 100644
index dc65d3e1..00000000
--- a/cli/src/format.rs
+++ /dev/null
@@ -1,205 +0,0 @@
-use crate::{Error, Result, Value};
-use stac::{Item, ItemCollection};
-use std::{
-    fs::File,
-    io::{BufRead, BufReader, Read, Write},
-    str::FromStr,
-};
-
-/// A file format that we can read and/or write.
-#[derive(Clone, Debug, PartialEq)]
-pub enum Format {
-    /// Pretty-printed JSON
-    PrettyJson,
-
-    /// Compact-printed JSON
-    CompactJson,
-
-    /// Streaming data.
-    ///
-    /// For JSON, this is newline-delimited JSON.
-    Streaming,
-
-    /// stac-geoparquet
-    #[cfg(feature = "geoparquet")]
-    Geoparquet(Option<parquet::basic::Compression>),
-}
-
-impl FromStr for Format {
-    type Err = Error;
-    fn from_str(s: &str) -> Result<Self> {
-        match s {
-            "json" | "pretty-json" | "pretty" => Ok(Self::PrettyJson),
-            "compact-json" | "compact" => Ok(Self::CompactJson),
-            "nd-json" | "ndjson" | "streaming" | "stream" => Ok(Self::Streaming),
-            // TODO we could embed compression in these strings
-            "geoparquet" | "parquet" | "stac-geoparquet" => {
-                #[cfg(feature = "geoparquet")]
-                {
-                    Ok(Self::Geoparquet(None))
-                }
-                #[cfg(not(feature = "geoparquet"))]
-                {
-                    tracing::error!(
-                        "geoparquet was requested, but the `geoparquet` feature is not enabled"
-                    );
-                    Err(Error::UnsupportedFormat(s.to_string()))
-                }
-            }
-            _ => Err(Error::UnsupportedFormat(s.to_string())),
-        }
-    }
-}
-
-impl Format {
-    /// Infers the format from the file extension.
-    ///
-    /// # Examples
-    ///
-    /// ```
-    /// use stac_cli::Format;
-    ///
-    /// assert_eq!(Format::infer("item.json").unwrap(), Format::CompactJson);
-    /// ```
-    pub fn infer(href: &str) -> Option<Format> {
-        href.rsplit_once('.').and_then(|(_, ext)| match ext {
-            "json" | "geojson" => Some(Format::CompactJson),
-            "parquet" | "geoparquet" => {
-                #[cfg(feature = "geoparquet")]
-                {
-                    Some(Format::Geoparquet(None))
-                }
-                #[cfg(not(feature = "geoparquet"))]
-                {
-                    tracing::warn!("'{}' has a geoparquet file extension, but the geoparquet feature is not enabled — falling back to compact json", href);
-                    Some(Format::CompactJson)
-                }
-            }
-            _ => None
-        })
-    }
-
-    /// Writes a [Value] to a [Write].
-    ///
-    /// # Examples
-    ///
-    /// ```
-    /// use stac_cli::Format;
-    ///
-    /// let item: stac::Value = stac::read("examples/simple-item.json").unwrap();
-    /// Format::PrettyJson.to_writer(std::io::stdout(), item).unwrap();
-    /// ```
-    pub fn to_writer(&self, mut writer: impl Write + Send, value: impl Into<Value>) -> Result<()> {
-        match self {
-            Format::PrettyJson => {
-                serde_json::to_writer_pretty(writer, &value.into()).map_err(Error::from)
-            }
-            Format::CompactJson => {
-                serde_json::to_writer(writer, &value.into()).map_err(Error::from)
-            }
-            Format::Streaming => {
-                let value = value.into();
-                match value {
-                    Value::Stac(value) => {
-                        if let stac::Value::ItemCollection(item_collection) = value {
-                            for item in item_collection {
-                                serde_json::to_writer(&mut writer, &item)?;
-                                writeln!(&mut writer)?;
-                            }
-                            Ok(())
-                        } else {
-                            serde_json::to_writer(&mut writer, &value)?;
-                            writeln!(writer).map_err(Error::from)
-                        }
-                    }
-                    Value::Json(value) => {
-                        if let serde_json::Value::Array(array) = value {
-                            for value in array {
-                                serde_json::to_writer(&mut writer, &value)?;
-                                writeln!(&mut writer)?;
-                            }
-                            Ok(())
-                        } else {
-                            serde_json::to_writer(&mut writer, &value)?;
-                            writeln!(writer).map_err(Error::from)
-                        }
-                    }
-                    Value::String(string) => writeln!(writer, "{}", string).map_err(Error::from),
-                }
-            }
-            #[cfg(feature = "geoparquet")]
-            Format::Geoparquet(compression) => {
-                let value = stac::Value::try_from(value.into())?;
-                let mut options = geoarrow::io::parquet::GeoParquetWriterOptions::default();
-                if let Some(compression) = compression {
-                    let writer_properties = parquet::file::properties::WriterProperties::builder()
-                        .set_compression(*compression)
-                        .build();
-                    options.writer_properties = Some(writer_properties);
-                }
-                stac::geoparquet::to_writer_with_options(writer, value, &options)
-                    .map_err(Error::from)
-            }
-        }
-    }
-
-    /// Reads a [stac::Value] from a file.
-    ///
-    /// # Examples
-    ///
-    /// ```
-    /// use stac_cli::Format;
-    ///
-    /// let item = Format::CompactJson.from_file("examples/simple-item.json").unwrap();
-    /// ```
-    pub fn from_file(&self, file: &str) -> Result<stac::Value> {
-        match self {
-            Format::CompactJson | Format::PrettyJson => stac::json::read(file).map_err(Error::from),
-            Format::Streaming => {
-                let reader = BufReader::new(File::open(file)?);
-                self.from_reader(reader)
-            }
-            #[cfg(feature = "geoparquet")]
-            Format::Geoparquet(_) => stac::geoparquet::read(file)
-                .map(stac::Value::from)
-                .map_err(Error::from),
-        }
-    }
-
-    /// Reads a [stac::Value] from a [Read].
-    ///
-    /// # Examples
-    ///
-    /// ```
-    /// use stac_cli::Format;
-    ///
-    /// let item = Format::CompactJson.from_file("examples/simple-item.json").unwrap();
-    /// ```
-    #[cfg_attr(not(feature = "geoparquet"), allow(unused_mut))]
-    pub fn from_reader<R: Read>(&self, mut reader: R) -> Result<stac::Value> {
-        match self {
-            Format::CompactJson | Format::PrettyJson => {
-                serde_json::from_reader(reader).map_err(Error::from)
-            }
-            Format::Streaming => {
-                let mut items = Vec::new();
-                for line in BufReader::new(reader).lines() {
-                    let item: Item = serde_json::from_str(&line?)?;
-                    items.push(item);
-                }
-                Ok(ItemCollection::from(items).into())
-            }
-            #[cfg(feature = "geoparquet")]
-            Format::Geoparquet(_) => {
-                use bytes::Bytes;
-
-                let mut buf = Vec::new();
-                let _ = reader.read_to_end(&mut buf)?;
-                let bytes = Bytes::from(buf);
-                stac::geoparquet::from_reader(bytes)
-                    .map(stac::Value::from)
-                    .map_err(Error::from)
-            }
-        }
-    }
-}
diff --git a/cli/src/input.rs b/cli/src/input.rs
new file mode 100644
index 00000000..e97797f3
--- /dev/null
+++ b/cli/src/input.rs
@@ -0,0 +1,120 @@
+use crate::{config::Config, Error, Result};
+use object_store::{local::LocalFileSystem, path::Path, ObjectStore};
+use stac::{Format, Item, ItemCollection, Value};
+use std::io::BufReader;
+use url::Url;
+
+/// The input to a CLI run.
+#[derive(Debug, Default)]
+pub(crate) struct Input {
+    format: Format,
+    reader: Reader,
+    config: Config,
+}
+
+#[derive(Debug, Default)]
+enum Reader {
+    ObjectStore {
+        object_store: Box<dyn ObjectStore>,
+        path: Path,
+    },
+    #[default]
+    Stdin,
+}
+
+impl Input {
+    /// Creates a new input.
+    pub(crate) fn new(
+        infile: impl Into<Option<String>>,
+        format: impl Into<Option<Format>>,
+        config: impl Into<Config>,
+    ) -> Result<Input> {
+        let infile = infile
+            .into()
+            .and_then(|infile| if infile == "-" { None } else { Some(infile) });
+        let format = format
+            .into()
+            .or_else(|| infile.as_deref().and_then(Format::infer_from_href))
+            .unwrap_or_default();
+        let config = config.into();
+        let reader = if let Some(infile) = infile {
+            let (object_store, path) = parse_href_opts(&infile, config.iter())?;
+            Reader::ObjectStore { object_store, path }
+        } else {
+            Reader::Stdin
+        };
+        Ok(Input {
+            format,
+            reader,
+            config,
+        })
+    }
+
+    /// Creates a new input with the given href.
+    pub(crate) fn with_href(&self, href: &str) -> Result<Input> {
+        let (object_store, path) = parse_href_opts(&href, self.config.iter())?;
+        let reader = Reader::ObjectStore { object_store, path };
+        Ok(Input {
+            format: self.format,
+            reader,
+            config: self.config.clone(),
+        })
+    }
+
+    /// Gets a STAC value from the input.
+    pub(crate) async fn get(&self) -> Result<Value> {
+        tracing::debug!("getting {}", self.format);
+        match &self.reader {
+            Reader::ObjectStore { object_store, path } => {
+                let bytes = object_store.get(&path).await?.bytes().await?;
+                match self.format {
+                    Format::Json => serde_json::from_slice(&bytes).map_err(Error::from),
+                    Format::NdJson => bytes
+                        .split(|c| *c == b'\n')
+                        .map(|line| serde_json::from_slice::<Item>(line).map_err(Error::from))
+                        .collect::<Result<Vec<_>>>()
+                        .map(ItemCollection::from)
+                        .map(Value::from),
+                    #[cfg(feature = "geoparquet")]
+                    Format::Geoparquet => stac::geoparquet::from_reader(bytes)
+                        .map(Value::from)
+                        .map_err(Error::from),
+                }
+            }
+            Reader::Stdin => match self.format {
+                Format::Json => serde_json::from_reader(std::io::stdin()).map_err(Error::from),
+                Format::NdJson => stac::ndjson::from_buf_reader(BufReader::new(std::io::stdin()))
+                    .map(Value::from)
+                    .map_err(Error::from),
+                #[cfg(feature = "geoparquet")]
+                Format::Geoparquet => {
+                    use std::io::Read;
+
+                    let mut buf = Vec::new();
+                    let _ = std::io::stdin().read_to_end(&mut buf)?;
+                    stac::geoparquet::from_reader(bytes::Bytes::from(buf))
+                        .map(Value::from)
+                        .map_err(Error::from)
+                }
+            },
+        }
+    }
+}
+
+pub(crate) fn parse_href_opts<I, K, V>(
+    href: &str,
+    options: I,
+) -> Result<(Box<dyn ObjectStore>, Path)>
+where
+    I: IntoIterator<Item = (K, V)>,
+    K: AsRef<str>,
+    V: Into<String>,
+{
+    if let Some(url) = Url::parse(href).ok() {
+        object_store::parse_url_opts(&url, options).map_err(Error::from)
+    } else {
+        let path = Path::from_filesystem_path(href)?;
+        let object_store = LocalFileSystem::new();
+        Ok((Box::new(object_store), path))
+    }
+}
diff --git a/cli/src/lib.rs b/cli/src/lib.rs
index f9f1c26d..c4a5dc8b 100644
--- a/cli/src/lib.rs
+++ b/cli/src/lib.rs
@@ -1,7 +1,19 @@
 //! STAC command-line interface (CLI).
+//!
+//! # Examples
+//!
+//! ```no_run
+//! use stac_cli::Args;
+//! use clap::Parser;
+//!
+//! let args = Args::parse_from(["stacrs", "--help"]);
+//! # tokio_test::block_on(async {
+//! args.run().await.unwrap();
+//! # })
+//! ```
 
 #![cfg_attr(docsrs, feature(doc_auto_cfg))]
-#![deny(
+#![warn(
     elided_lifetimes_in_paths,
     explicit_outlives_requirements,
     keyword_idents,
@@ -31,14 +43,16 @@
     warnings
 )]
 
-pub mod args;
+mod args;
+mod config;
 mod error;
-mod format;
+mod input;
+mod output;
 #[cfg(feature = "python")]
 mod python;
 mod value;
 
-pub use {args::Args, error::Error, format::Format, value::Value};
+pub use {args::Args, error::Error, value::Value};
 
 /// Crate-specific result type.
 pub type Result<T> = std::result::Result<T, Error>;
diff --git a/cli/src/main.rs b/cli/src/main.rs
index 100391b9..00ce6f4c 100644
--- a/cli/src/main.rs
+++ b/cli/src/main.rs
@@ -7,7 +7,7 @@ async fn main() {
     tracing_subscriber::fmt()
         .with_max_level(args.log_level())
         .init();
-    std::process::exit(match args.run(std::io::stdout()).await {
+    std::process::exit(match args.run().await {
         Ok(()) => 0,
         Err(err) => {
             eprintln!("ERROR: {}", err);
diff --git a/cli/src/output.rs b/cli/src/output.rs
new file mode 100644
index 00000000..febdf0d7
--- /dev/null
+++ b/cli/src/output.rs
@@ -0,0 +1,182 @@
+//! Structures for writing output data.
+
+use crate::{config::Config, value::Value, Error, Result};
+use object_store::{buffered::BufWriter, path::Path, ObjectStore, PutPayload, PutResult};
+use std::{fmt::Display, io::IsTerminal, pin::Pin, str::FromStr, sync::Arc};
+use tokio::{
+    fs::File,
+    io::{AsyncWrite, AsyncWriteExt},
+};
+use url::Url;
+
+/// The output from a CLI run.
+#[allow(missing_debug_implementations)]
+pub(crate) struct Output {
+    /// The output format.
+    pub format: Format,
+
+    writer: Writer,
+}
+
+/// The output format.
+#[derive(Clone, Debug, PartialEq, Copy)]
+pub(crate) enum Format {
+    /// Pretty-printed JSON, good for terminals.
+    PrettyJson,
+    /// Compact JSON, good for files.
+    CompactJson,
+    /// Newline-delimited JSON
+    NdJson,
+    /// stac-geoparquet
+    #[cfg(feature = "geoparquet")]
+    Geoparquet(parquet::basic::Compression),
+}
+
+struct Writer {
+    stream: Pin<Box<dyn AsyncWrite + Send>>,
+    object_store: Option<(Arc<Box<dyn ObjectStore>>, Path)>,
+}
+
+impl Output {
+    /// Creates a new output from an optional outfile and an optional format.
+    pub(crate) async fn new(
+        outfile: Option<String>,
+        format: Option<Format>,
+        config: impl Into<Config>,
+        create_parent_directories: bool,
+    ) -> Result<Output> {
+        let format = format
+            .or_else(|| outfile.as_deref().and_then(Format::infer_from_href))
+            .unwrap_or_else(|| {
+                if outfile.is_some() && !std::io::stdout().is_terminal() {
+                    Format::CompactJson
+                } else {
+                    Format::PrettyJson
+                }
+            });
+        let writer = if let Some(outfile) = outfile {
+            if let Some(url) = Url::parse(&outfile).ok().and_then(|url| {
+                if url.scheme() == "file" {
+                    None
+                } else {
+                    Some(url)
+                }
+            }) {
+                let (object_store, path) =
+                    object_store::parse_url_opts(&url, config.into().iter())?;
+                let object_store = Arc::new(object_store);
+                let stream = BufWriter::new(object_store.clone(), path.clone());
+                Writer {
+                    stream: Box::pin(stream),
+                    object_store: Some((object_store, path)),
+                }
+            } else {
+                if create_parent_directories {
+                    if let Some(parent) = std::path::Path::new(&outfile).parent() {
+                        std::fs::create_dir_all(parent)?;
+                    }
+                }
+                Writer {
+                    stream: Box::pin(File::create(outfile).await?),
+                    object_store: None,
+                }
+            }
+        } else {
+            Writer {
+                stream: Box::pin(tokio::io::stdout()),
+                object_store: None,
+            }
+        };
+        Ok(Output { format, writer })
+    }
+
+    /// Streams a value to the output
+    pub(crate) async fn stream(&mut self, value: Value) -> Result<()> {
+        let bytes = value.into_bytes(Format::NdJson)?;
+        self.writer.stream.write_all(&bytes).await?;
+        self.writer.stream.flush().await?;
+        Ok(())
+    }
+
+    /// Puts a value to the output.
+    pub(crate) async fn put(&mut self, value: Value) -> Result<Option<PutResult>> {
+        let bytes = value.into_bytes(self.format)?;
+        if let Some((object_store, path)) = &self.writer.object_store {
+            object_store
+                .put(path, PutPayload::from_bytes(bytes.into()))
+                .await
+                .map(Some)
+                .map_err(Error::from)
+        } else {
+            let output = self.writer.stream.write_all(&bytes).await.map(|_| None)?;
+            self.writer.stream.flush().await?;
+            Ok(output)
+        }
+    }
+}
+
+impl FromStr for Format {
+    type Err = Error;
+    fn from_str(s: &str) -> Result<Format> {
+        let s = s.to_ascii_lowercase();
+        match s.as_str() {
+            "json" | "geojson" => Ok(Self::CompactJson),
+            "ndjson" => Ok(Self::NdJson),
+            _ => {
+                #[cfg(feature = "geoparquet")]
+                if s.starts_with("parquet") || s.starts_with("geoparquet") {
+                    if let Some((_, compression)) = s.split_once('[') {
+                        if let Some(stop) = compression.find(']') {
+                            Ok(Self::Geoparquet(compression[..stop].parse()?))
+                        } else {
+                            Err(Error::UnsupportedFormat(s.to_string()))
+                        }
+                    } else {
+                        Ok(Self::Geoparquet(parquet::basic::Compression::UNCOMPRESSED))
+                    }
+                } else {
+                    Err(Error::UnsupportedFormat(s.to_string()))
+                }
+                #[cfg(not(feature = "geoparquet"))]
+                Err(Error::UnsupportedFormat(s.to_string()))
+            }
+        }
+    }
+}
+
+impl Display for Format {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        match self {
+            Format::PrettyJson => f.write_str("pretty-json"),
+            Format::CompactJson => f.write_str("compact-json"),
+            Format::NdJson => f.write_str("nd-json"),
+            #[cfg(feature = "geoparquet")]
+            Format::Geoparquet(compression) => {
+                if *compression == parquet::basic::Compression::UNCOMPRESSED {
+                    f.write_str("geoparquet")
+                } else {
+                    write!(f, "geoparquet[{}]", compression)
+                }
+            }
+        }
+    }
+}
+
+impl Format {
+    fn infer_from_href(href: &str) -> Option<Format> {
+        href.rsplit_once('.').and_then(|(_, ext)| ext.parse().ok())
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    #[test]
+    #[cfg(feature = "geoparquet")]
+    fn geoparquet_compression() {
+        let format: super::Format = "geoparquet[snappy]".parse().unwrap();
+        assert_eq!(
+            format,
+            super::Format::Geoparquet(parquet::basic::Compression::SNAPPY)
+        );
+    }
+}
diff --git a/cli/src/python.rs b/cli/src/python.rs
index e39bf4ac..a861a489 100644
--- a/cli/src/python.rs
+++ b/cli/src/python.rs
@@ -18,7 +18,7 @@ fn main() -> PyResult<i64> {
             .unwrap()
             .block_on(async {
                 // We skip one because the first argument is going to be the python interpreter.
-                match args.run(std::io::stdout()).await {
+                match args.run().await {
                     Ok(()) => 0,
                     Err(err) => {
                         eprintln!("ERROR: {}", err);
diff --git a/cli/src/value.rs b/cli/src/value.rs
index 4a4a28b4..55dbd5dd 100644
--- a/cli/src/value.rs
+++ b/cli/src/value.rs
@@ -1,6 +1,8 @@
+use crate::{output::Format, Error, Result};
+use bytes::Bytes;
 use serde::Serialize;
 
-/// An output value, which can either be a [serde_json::Value], a [stac::Value], or a [String].
+/// An output value, which can either be a [serde_json::Value] or a [stac::Value].
 #[derive(Debug, Serialize)]
 #[serde(untagged)]
 pub enum Value {
@@ -9,9 +11,80 @@ pub enum Value {
 
     /// A JSON value.
     Json(serde_json::Value),
+}
+
+impl Value {
+    pub(crate) fn into_bytes(self, format: Format) -> Result<Bytes> {
+        match self {
+            Self::Stac(value) => match format {
+                Format::CompactJson => serde_json::to_vec(&value)
+                    .map(Bytes::from)
+                    .map_err(Error::from),
+                Format::PrettyJson => serde_json::to_vec_pretty(&value)
+                    .map(Bytes::from)
+                    .map_err(Error::from),
+                Format::NdJson => {
+                    if let stac::Value::ItemCollection(item_collection) = value {
+                        let mut buf = Vec::new();
+                        for item in item_collection.items {
+                            serde_json::to_writer(&mut buf, &item)?;
+                            buf.push(b'\n');
+                        }
+                        Ok(buf.into())
+                    } else {
+                        serde_json::to_vec(&value)
+                            .map(Bytes::from)
+                            .map_err(Error::from)
+                    }
+                }
+                #[cfg(feature = "geoparquet")]
+                Format::Geoparquet(compression) => geoparquet_bytes(value, compression),
+            },
+            Self::Json(value) => match format {
+                Format::CompactJson => serde_json::to_vec(&value)
+                    .map(Bytes::from)
+                    .map_err(Error::from),
+                Format::PrettyJson => serde_json::to_vec_pretty(&value)
+                    .map(Bytes::from)
+                    .map_err(Error::from),
+                Format::NdJson => {
+                    if let serde_json::Value::Array(array) = value {
+                        let mut buf = Vec::new();
+                        for value in array {
+                            serde_json::to_writer(&mut buf, &value)?;
+                            buf.push(b'\n');
+                        }
+                        Ok(buf.into())
+                    } else {
+                        serde_json::to_vec(&value)
+                            .map(Bytes::from)
+                            .map_err(Error::from)
+                    }
+                }
+                #[cfg(feature = "geoparquet")]
+                Format::Geoparquet(compression) => {
+                    geoparquet_bytes(serde_json::from_value(value)?, compression)
+                }
+            },
+        }
+    }
+}
 
-    /// A string value.
-    String(String),
+#[cfg(feature = "geoparquet")]
+fn geoparquet_bytes(value: stac::Value, compression: parquet::basic::Compression) -> Result<Bytes> {
+    tracing::debug!(
+        "converting STAC {} to geoparquet bytes using {} compression",
+        value.type_name(),
+        compression
+    );
+    let mut options = geoarrow::io::parquet::GeoParquetWriterOptions::default();
+    let writer_properties = parquet::file::properties::WriterProperties::builder()
+        .set_compression(compression)
+        .build();
+    options.writer_properties = Some(writer_properties);
+    let mut bytes = Vec::new();
+    stac::geoparquet::to_writer_with_options(&mut bytes, value, &options)?;
+    Ok(bytes.into())
 }
 
 impl From<stac::Value> for Value {
@@ -32,19 +105,12 @@ impl From<serde_json::Value> for Value {
     }
 }
 
-impl From<String> for Value {
-    fn from(value: String) -> Self {
-        Self::String(value)
-    }
-}
-
 impl TryFrom<Value> for stac::Value {
-    type Error = serde_json::Error;
-    fn try_from(value: Value) -> Result<Self, Self::Error> {
+    type Error = Error;
+    fn try_from(value: Value) -> Result<Self> {
         match value {
             Value::Stac(value) => Ok(value),
-            Value::Json(value) => serde_json::from_value(value),
-            Value::String(string) => serde_json::from_str(&string),
+            Value::Json(value) => serde_json::from_value(value).map_err(Error::from),
         }
     }
 }
diff --git a/core/CHANGELOG.md b/core/CHANGELOG.md
index d610a140..f8c1a8d1 100644
--- a/core/CHANGELOG.md
+++ b/core/CHANGELOG.md
@@ -10,6 +10,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 
 - 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))
 
 ### Changed
 
diff --git a/core/src/error.rs b/core/src/error.rs
index e9d978db..fdac7c6b 100644
--- a/core/src/error.rs
+++ b/core/src/error.rs
@@ -133,6 +133,10 @@ pub enum Error {
     #[error("unsupported migration: {0} to {1}")]
     UnsupportedMigration(Version, Version),
 
+    /// Unsupported file format.
+    #[error("unsupported format: {0}")]
+    UnsupportedFormat(String),
+
     /// [url::ParseError]
     #[error(transparent)]
     Url(#[from] url::ParseError),
diff --git a/core/src/format.rs b/core/src/format.rs
new file mode 100644
index 00000000..ee0ba17e
--- /dev/null
+++ b/core/src/format.rs
@@ -0,0 +1,68 @@
+use crate::{Error, Result};
+use std::{fmt::Display, str::FromStr};
+
+/// The format of STAC data.
+#[derive(Debug, Clone, Copy, Default, PartialEq)]
+pub enum Format {
+    /// JSON data (the default).
+    #[default]
+    Json,
+
+    /// Newline-delimited JSON.
+    NdJson,
+
+    /// [stac-geoparquet](https://github.com/stac-utils/stac-geoparquet)
+    #[cfg(feature = "geoparquet")]
+    Geoparquet,
+}
+
+impl Format {
+    /// Infer a format from a file extension.
+    ///
+    /// # Examples
+    ///
+    /// ```
+    /// use stac::Format;
+    ///
+    /// assert_eq!(Format::Json, Format::infer_from_href("item.json").unwrap());
+    /// ```
+    pub fn infer_from_href(href: &str) -> Option<Format> {
+        href.rsplit_once('.').and_then(|(_, ext)| ext.parse().ok())
+    }
+}
+
+impl Display for Format {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.write_str(match self {
+            Self::Json => "json",
+            Self::NdJson => "ndjson",
+            #[cfg(feature = "geoparquet")]
+            Self::Geoparquet => "geoparquet",
+        })
+    }
+}
+
+impl FromStr for Format {
+    type Err = Error;
+
+    fn from_str(s: &str) -> Result<Format> {
+        match s.to_ascii_lowercase().as_str() {
+            "json" | "geojson" => Ok(Self::Json),
+            "ndjson" => Ok(Self::NdJson),
+            #[cfg(feature = "geoparquet")]
+            "geoparquet" | "parquet" => Ok(Self::Geoparquet),
+            _ => Err(Error::UnsupportedFormat(s.to_string())),
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::Format;
+
+    #[test]
+    #[cfg(feature = "geoparquet")]
+    fn parse_geoparquet() {
+        assert_eq!("parquet".parse::<Format>().unwrap(), Format::Geoparquet);
+    }
+}
diff --git a/core/src/lib.rs b/core/src/lib.rs
index 424fafbc..7fba3bec 100644
--- a/core/src/lib.rs
+++ b/core/src/lib.rs
@@ -143,6 +143,7 @@ pub mod datetime;
 mod error;
 pub mod extensions;
 mod fields;
+mod format;
 #[cfg(feature = "gdal")]
 pub mod gdal;
 #[cfg(feature = "geo")]
@@ -175,6 +176,7 @@ pub use {
     error::Error,
     extensions::{Extension, Extensions},
     fields::Fields,
+    format::Format,
     href::{href_to_url, Href},
     io::read,
     item::{FlatItem, Item, Properties, ITEM_TYPE},


Test patch:
diff --git a/cli/tests/help.rs b/cli/tests/help.rs
deleted file mode 100644
index ac5d77ae..00000000
--- a/cli/tests/help.rs
+++ /dev/null
@@ -1,7 +0,0 @@
-use assert_cmd::Command;
-
-#[test]
-fn help() {
-    let mut command = Command::cargo_bin("stacrs").unwrap();
-    command.arg("help").assert().success();
-}
diff --git a/cli/tests/item.rs b/cli/tests/item.rs
deleted file mode 100644
index e98f91cb..00000000
--- a/cli/tests/item.rs
+++ /dev/null
@@ -1,7 +0,0 @@
-use assert_cmd::Command;
-
-#[test]
-fn item() {
-    let mut command = Command::cargo_bin("stacrs").unwrap();
-    command.arg("item").arg("an-id").assert().success();
-}
diff --git a/cli/tests/items.rs b/cli/tests/items.rs
deleted file mode 100644
index 774a43ca..00000000
--- a/cli/tests/items.rs
+++ /dev/null
@@ -1,14 +0,0 @@
-use assert_cmd::Command;
-use stac::ItemCollection;
-
-#[test]
-fn items() {
-    let mut command = Command::cargo_bin("stacrs").unwrap();
-    let output = command
-        .arg("items")
-        .arg("../core/assets/dataset_geo.tif")
-        .arg("../core/assets/dataset.tif")
-        .unwrap();
-    let item_collection: ItemCollection = serde_json::from_slice(&output.stdout).unwrap();
-    assert_eq!(item_collection.items.len(), 2);
-}
diff --git a/cli/tests/smoke.rs b/cli/tests/smoke.rs
new file mode 100644
index 00000000..03b0c6ea
--- /dev/null
+++ b/cli/tests/smoke.rs
@@ -0,0 +1,72 @@
+use assert_cmd::Command;
+use stac::ItemCollection;
+use std::{fs::File, io::Read};
+
+#[test]
+fn help() {
+    let mut command = Command::cargo_bin("stacrs").unwrap();
+    command.arg("help").assert().success();
+}
+
+#[test]
+fn item() {
+    let mut command = Command::cargo_bin("stacrs").unwrap();
+    command.arg("item").arg("an-id").assert().success();
+}
+
+#[test]
+fn items() {
+    let mut command = Command::cargo_bin("stacrs").unwrap();
+    let output = command
+        .arg("items")
+        .arg("../core/assets/dataset_geo.tif")
+        .arg("../core/assets/dataset.tif")
+        .unwrap();
+    let item_collection: ItemCollection = serde_json::from_slice(&output.stdout).unwrap();
+    assert_eq!(item_collection.items.len(), 2);
+}
+
+#[test]
+fn migrate() {
+    let mut command = Command::cargo_bin("stacrs").unwrap();
+    command
+        .arg("migrate")
+        .arg("examples/simple-item.json")
+        .arg("--version")
+        .arg("1.1.0-beta.1")
+        .assert()
+        .success();
+}
+
+#[test]
+fn translate() {
+    let mut command = Command::cargo_bin("stacrs").unwrap();
+    command
+        .arg("translate")
+        .arg("examples/simple-item.json")
+        .arg("-o")
+        .arg("ndjson")
+        .assert()
+        .success();
+}
+
+#[test]
+fn validate() {
+    let mut command = Command::cargo_bin("stacrs").unwrap();
+    command
+        .arg("validate")
+        .arg("examples/simple-item.json")
+        .assert()
+        .success();
+}
+
+#[test]
+fn validate_stdin() {
+    let mut command = Command::cargo_bin("stacrs").unwrap();
+    let mut item = String::new();
+    File::open("examples/simple-item.json")
+        .unwrap()
+        .read_to_string(&mut item)
+        .unwrap();
+    command.arg("validate").write_stdin(item).assert().success();
+}
diff --git a/cli/tests/validate.rs b/cli/tests/validate.rs
deleted file mode 100644
index c9ae9218..00000000
--- a/cli/tests/validate.rs
+++ /dev/null
@@ -1,23 +0,0 @@
-use assert_cmd::Command;
-use std::{fs::File, io::Read};
-
-#[test]
-fn validate() {
-    let mut command = Command::cargo_bin("stacrs").unwrap();
-    command
-        .arg("validate")
-        .arg("examples/simple-item.json")
-        .assert()
-        .success();
-}
-
-#[test]
-fn validate_stdin() {
-    let mut command = Command::cargo_bin("stacrs").unwrap();
-    let mut item = String::new();
-    File::open("examples/simple-item.json")
-        .unwrap()
-        .read_to_string(&mut item)
-        .unwrap();
-    command.arg("validate").write_stdin(item).assert().success();
-}