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

Issue:
Allow a file input to --intersects



Discussion / hints:

Fix patch:
diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
index 6b04891b..823758a5 100644
--- a/.github/workflows/ci.yml
+++ b/.github/workflows/ci.yml
@@ -30,7 +30,7 @@ jobs:
           - "-p stac-async"
           - "-p stac-cli --no-default-features"
           - "-p stac-cli --no-default-features -F geoparquet"
-          - "-p stac-duckdb -F bundled"
+          - "-p stac-duckdb -F duckdb/bundled"
           - "-p stac-server"
           - "-p stac-server -F axum"
           - "-p stac-validate"
@@ -50,7 +50,7 @@ jobs:
           sudo apt-get update
           sudo apt-get install libgdal-dev
       - name: Test
-        run: cargo test -p stac -p stac-cli --all-features
+        run: cargo test -p stac -p stac-duckdb -p stac-cli --all-features -F duckdb/bundled
   test-ubuntu-with-pgstac:
     runs-on: ubuntu-latest
     services:
diff --git a/cli/Cargo.toml b/cli/Cargo.toml
index e62b7049..3a33d69b 100644
--- a/cli/Cargo.toml
+++ b/cli/Cargo.toml
@@ -9,6 +9,7 @@ repository = "https://github.com/stac-utils/stac-rs"
 license = "MIT OR Apache-2.0"
 keywords = ["geospatial", "stac", "metadata", "geo", "raster"]
 categories = ["science", "data-structures"]
+rust-version = "1.75"
 
 [features]
 default = ["gdal", "geoparquet", "pgstac"]
@@ -21,7 +22,7 @@ geoparquet = [
     "parquet",
 ]
 pgstac = ["stac-server/pgstac"]
-python = ["dep:pyo3", "pgstac", "duckdb", "stac-duckdb/bundled", "geoparquet"]
+python = ["dep:pyo3", "pgstac", "geoparquet"]
 
 [dependencies]
 axum = "0.7"
@@ -47,7 +48,6 @@ tokio = { version = "1.23", features = ["macros", "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/README.md b/cli/README.md
index 3f156afe..3b388636 100644
--- a/cli/README.md
+++ b/cli/README.md
@@ -17,6 +17,9 @@ cargo install stac-cli
 Or:
 
 ```shell
+# NOTE: The version from PyPI does not include GDAL or DuckDB support. If you
+# need to use these features, install via `cargo install` (GDAL is enabled by
+# default) or `cargo install -F duckdb` (DuckDB is not).
 pip install stacrs-cli
 ```
 
@@ -26,17 +29,14 @@ Then:
 stacrs --help
 ```
 
-**NOTE:** the version from PyPI (installed with `pip`) does not include GDAL support.
-If you need to use [gdal](../core/README.md#gdal) features, install via `cargo install`.
-
 ## Usage
 
 **stacrs** provides the following subcommands:
 
 - `stacrs item`: create STAC items and combine them into item collections
-- `stacrs search`: search STAC APIs
-- `stacrs serve`: serve a STAC API
-- `stacrs sort`: sort the fields of STAC items, catalogs, and collections
+- `stacrs migrate`: migrate a STAC object to another version
+- `stacrs search`: search STAC APIs (and geoparquet, with the experimental `duckdb` feature)
+- `stacrs serve`: serve a STAC API (optionally, with a [pgstac](https://github.com/stac-utils/pgstac) backend)
 - `stacrs translate`: convert STAC values from one format to another
 - `stacrs validate`: validate STAC items, catalogs, and collections using [json-schema](https://json-schema.org/)
 
@@ -44,8 +44,15 @@ Use the `--help` flag to see all available options for the CLI and the subcomman
 
 ## Features
 
-By default, the CLI builds w/ [GDAL](https://gdal.org) support, and will error if GDAL is not installed on your system.
-If you don't want to use GDAL, you can disable the default features:
+This crate has five features, three of them on by default:
+
+- `duckdb`: experimental support for querying [stac-geoparquet](https://github.com/stac-utils/stac-geoparquet) files using [DuckDB](https://duckdb.org/)
+- `gdal`: read geospatial data from rasters (enabled by default)
+- `geoparquet`: read and write [stac-geoparquet](https://github.com/stac-utils/stac-geoparquet) (enabled by default)
+- `pgstac`: enable a [pgstac](https://github.com/stac-utils/pgstac) backend for `stacrs serve` (enabled by default)
+- `python`: create an entrypoint that can be called from Python (used to enable `pip install stacrs-cli`)
+
+If you don't want to use GDAL or any of the other default features:
 
 ```shell
 cargo install stac-cli --no-default-features
diff --git a/cli/src/args.rs b/cli/src/args.rs
deleted file mode 100644
index 8536d4ad..00000000
--- a/cli/src/args.rs
+++ /dev/null
@@ -1,270 +0,0 @@
-use crate::{Format, Result, Subcommand};
-use clap::Parser;
-use stac::{Version, STAC_VERSION};
-use std::{
-    fs::File,
-    io::{BufWriter, Write},
-};
-
-/// CLI arguments.
-#[derive(Parser, Debug)]
-#[command(author, version, about, long_about = None)]
-pub struct Args {
-    /// Use a compact representation of the output, if possible.
-    #[arg(short, long)]
-    pub compact: bool,
-
-    /// The input format. If not provided, the format will be detected from the input file extension when possible.
-    #[arg(short, long)]
-    pub input_format: Option<Format>,
-
-    /// The output format. If not provided, the format will be detected from the output file extension when possible.
-    #[arg(short, long, value_enum)]
-    pub output_format: Option<Format>,
-
-    /// The type of geoparquet compression to use.
-    ///
-    /// Possible values: uncompressed (default), snappy, gzip(level), lzo, brotli(level), lz4, zstd(level), lz4_raw
-    #[arg(long, value_enum)]
-    #[cfg(feature = "geoparquet")]
-    pub geoparquet_compression: Option<parquet::basic::Compression>,
-
-    /// The subcommand to run.
-    #[command(subcommand)]
-    pub subcommand: Subcommand,
-}
-
-/// Arguments for creating an item.
-#[derive(clap::Args, Debug)]
-pub struct ItemArgs {
-    /// The item id or asset href.
-    pub id_or_href: String,
-
-    /// The item id, if the positional argument is an href.
-    ///
-    /// If not provided, will be inferred from the filename in the href.
-    #[arg(short, long)]
-    pub id: Option<String>,
-
-    /// The asset key, if the positional argument is an href.
-    #[arg(short, long, default_value = "data")]
-    pub key: String,
-
-    /// The asset roles, if the positional argument is an href.
-    ///
-    /// Can be provided multiple times.
-    #[arg(short, long)]
-    pub role: Vec<String>,
-
-    /// Allow relative paths.
-    ///
-    /// If false, all path will be canonicalized, which requires that the
-    /// files actually exist on the filesystem.
-    #[arg(long)]
-    pub allow_relative_paths: bool,
-
-    /// Don't use GDAL for item creation, if the positional argument is an href.
-    ///
-    /// Automatically set to true if this crate is compiled without GDAL.
-    #[arg(long)]
-    pub disable_gdal: bool,
-
-    /// Collect an item or item collection from standard input, and add the
-    /// newly created to it into a new item collection.
-    #[arg(short, long)]
-    pub collect: bool,
-
-    /// The file to write the item to.
-    ///
-    /// If not provided, the item will be written to standard output.
-    pub outfile: Option<String>,
-}
-
-/// Arguments for searching a STAC API.
-#[derive(Debug, clap::Args)]
-pub struct SearchArgs {
-    /// The href of the STAC API.
-    pub href: String,
-
-    /// The maximum number of items to print.
-    #[arg(short, long)]
-    pub max_items: Option<usize>,
-
-    /// The maximum number of results to return (page size).
-    #[arg(short, long)]
-    pub limit: Option<String>,
-
-    /// Requested bounding box.
-    #[arg(short, long)]
-    pub bbox: Option<String>,
-
-    /// Requested bounding box.
-    ///
-    /// Use double dots `..` for open date ranges.
-    #[arg(short, long)]
-    pub datetime: Option<String>,
-
-    /// Searches items by performing intersection between their geometry and
-    /// provided GeoJSON geometry.
-    ///
-    /// All GeoJSON geometry types must be supported.
-    #[arg(long)]
-    pub intersects: Option<String>,
-
-    /// Comma-delimited list of one ore more Item ids to return.
-    #[arg(short, long)]
-    pub ids: Option<String>,
-
-    /// Comma-delimited list of one or more Collection IDs that each matching
-    /// Item must be in.
-    #[arg(short, long)]
-    pub collections: Option<String>,
-
-    /// Include/exclude fields from item collections.
-    #[arg(long)]
-    pub fields: Option<String>,
-
-    /// Fields by which to sort results.
-    #[arg(short, long)]
-    pub sortby: Option<String>,
-
-    /// Recommended to not be passed, but server must only accept
-    /// <http://www.opengis.net/def/crs/OGC/1.3/CRS84> as a valid value, may
-    /// reject any others
-    #[arg(long)]
-    pub filter_crs: Option<String>,
-
-    /// CQL2 filter expression.
-    #[arg(long)]
-    pub filter_lang: Option<String>,
-
-    /// CQL2 filter expression.
-    #[arg(short, long)]
-    pub filter: Option<String>,
-
-    /// Stream the items to output as ndjson.
-    #[arg(long)]
-    pub stream: bool,
-
-    /// Search with DuckDB instead of a STAC API client.
-    ///
-    /// ⚠ Experimental: this is a new feature whose behavior might change without warning.
-    ///
-    /// If true, or if not provided and the href's extension is `parquet` or `geoparquet`, use DuckDB. Set to `false` to always use a STAC API client.
-    #[arg(long)]
-    pub duckdb: Option<bool>,
-
-    /// The file to write the output to.
-    ///
-    /// If not provided, the output will be written to standard output.
-    pub outfile: Option<String>,
-}
-
-/// Arguments for serving a STAC API.
-#[derive(clap::Args, Debug)]
-pub struct ServeArgs {
-    /// Hrefs of STAC collections and items to load before starting the server.
-    pub href: Vec<String>,
-
-    /// The pgstac connection string.
-    #[arg(long)]
-    pub pgstac: Option<String>,
-
-    /// Don't auto-create collections for items that are missing them.
-    #[arg(long)]
-    pub dont_auto_create_collections: bool,
-
-    /// Don't follow links in collections to more items.
-    #[arg(long)]
-    pub dont_follow_links: bool,
-}
-
-/// Arguments for sorting a STAC value.
-#[derive(clap::Args, Debug)]
-pub struct SortArgs {
-    /// The href of the STAC to sort.
-    ///
-    /// If this is not provided, or is `-`, will read from standard input.
-    pub infile: Option<String>,
-
-    /// The output filename.
-    ///
-    /// If this is not provided, output will be printed to standard output.
-    pub outfile: Option<String>,
-}
-
-/// Arguments for validating a STAC value.
-#[derive(clap::Args, Debug)]
-pub struct ValidateArgs {
-    /// The href of the STAC object or endpoint.
-    ///
-    /// The validator will make some decisions depending on what type of
-    /// data is returned from the href. If it's a STAC Catalog, Collection,
-    /// or Item, that object will be validated. If its a collections
-    /// endpoint from a STAC API, all collections will be validated.
-    /// Additional behavior TBD.
-    ///
-    /// If this is not provided, or is `-`, will read from standard input.
-    pub href: Option<String>,
-}
-
-/// Arguments for translating STAC values.
-#[derive(clap::Args, Debug)]
-pub struct TranslateArgs {
-    /// The input STAC value.
-    ///
-    /// If this is not provided, or is `-`, input will be read from standard
-    /// input.
-    pub infile: Option<String>,
-
-    /// The output STAC value.
-    ///
-    /// If not provided, output will be printed to standard output.
-    pub outfile: Option<String>,
-}
-
-/// Arguments for migrating STAC values.
-#[derive(clap::Args, Debug)]
-pub struct MigrateArgs {
-    /// The input STAC value.
-    ///
-    /// If this is not provided, or is `-`, input will be read from standard
-    /// input.
-    pub infile: Option<String>,
-
-    /// The output STAC value.
-    ///
-    /// If not provided, output will be printed to standard output.
-    pub outfile: Option<String>,
-
-    /// The STAC version to migrate to.
-    #[arg(short, long, default_value_t = STAC_VERSION)]
-    pub version: Version,
-}
-
-impl Args {
-    pub(crate) fn writer(&self) -> Result<Box<dyn Write + Send>> {
-        if let Some(outfile) = self.subcommand.outfile() {
-            let file = File::create(outfile)?;
-            Ok(Box::new(BufWriter::new(file)))
-        } else {
-            Ok(Box::new(std::io::stdout()))
-        }
-    }
-
-    pub(crate) fn input_format(&self) -> Format {
-        self.input_format
-            .or_else(|| self.subcommand.infile().and_then(Format::maybe_from_href))
-            .unwrap_or_default()
-    }
-
-    pub(crate) fn output_format(&self) -> Format {
-        self.output_format
-            .or_else(|| self.subcommand.outfile().and_then(Format::maybe_from_href))
-            .unwrap_or_default()
-    }
-
-    pub(crate) fn outfile(&self) -> Option<&str> {
-        self.subcommand.outfile()
-    }
-}
diff --git a/cli/src/args/item.rs b/cli/src/args/item.rs
new file mode 100644
index 00000000..9d1c2a0e
--- /dev/null
+++ b/cli/src/args/item.rs
@@ -0,0 +1,68 @@
+use super::{Input, Run};
+use crate::{Result, Value};
+use stac::{item::Builder, Asset};
+use std::path::Path;
+use tokio::sync::mpsc::Sender;
+
+/// Arguments for the `item` subcommand.
+#[derive(clap::Args, Debug)]
+pub struct Args {
+    /// The item id or asset href
+    id_or_href: String,
+
+    /// The output file, if not provided the item will be printed to standard output
+    outfile: Option<String>,
+
+    /// The asset's key
+    #[arg(short, long, default_value = "data")]
+    key: String,
+
+    /// Roles to use for the created asset
+    #[arg(short, long = "role", default_values_t = ["data".to_string()])]
+    roles: Vec<String>,
+
+    /// Don't use GDAL to add geospatial metadata to the item
+    #[cfg(feature = "gdal")]
+    #[arg(long)]
+    disable_gdal: bool,
+
+    /// Allow assets to have relative hrefs
+    #[arg(long)]
+    allow_relative_hrefs: bool,
+}
+
+impl Run for Args {
+    async fn run(self, _: Input, _: 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 = id
+            .or_else(|| {
+                Path::new(href.as_ref().expect("if id is none, href should exist"))
+                    .file_stem()
+                    .map(|s| s.to_string_lossy().to_string())
+            })
+            .unwrap_or_else(|| "default".to_string());
+        let mut builder = Builder::new(id).canonicalize_paths(!self.allow_relative_hrefs);
+        #[cfg(feature = "gdal")]
+        {
+            builder = builder.enable_gdal(!self.disable_gdal);
+        }
+        if let Some(href) = href {
+            let mut asset = Asset::new(href);
+            asset.roles = self.roles;
+            builder = builder.asset(self.key, asset);
+        }
+        let item = builder.build()?;
+        Ok(Some(stac::Value::from(item).into()))
+    }
+
+    fn take_outfile(&mut self) -> Option<String> {
+        self.outfile.take()
+    }
+}
diff --git a/cli/src/args/migrate.rs b/cli/src/args/migrate.rs
new file mode 100644
index 00000000..fdcc60e1
--- /dev/null
+++ b/cli/src/args/migrate.rs
@@ -0,0 +1,33 @@
+use super::{Input, Run};
+use crate::{Error, Result, Value};
+use stac::{Migrate, Version, STAC_VERSION};
+use tokio::sync::mpsc::Sender;
+
+/// Arguments for the `migrate` subcommand.
+#[derive(clap::Args, Debug)]
+pub struct Args {
+    /// The input file, if not provided or `-` the input will be read from standard input
+    infile: Option<String>,
+
+    /// The output file, if not provided the item will be printed to standard output
+    outfile: Option<String>,
+
+    /// The version to migrate to
+    #[arg(long, default_value_t = STAC_VERSION)]
+    version: Version,
+}
+
+impl Run for Args {
+    async fn run(self, input: Input, _: Sender<Value>) -> Result<Option<Value>> {
+        let value = input.read(self.infile)?;
+        value
+            .migrate(self.version)
+            .map(Value::from)
+            .map(Some)
+            .map_err(Error::from)
+    }
+
+    fn take_outfile(&mut self) -> Option<String> {
+        self.outfile.take()
+    }
+}
diff --git a/cli/src/args/mod.rs b/cli/src/args/mod.rs
new file mode 100644
index 00000000..fa013e5e
--- /dev/null
+++ b/cli/src/args/mod.rs
@@ -0,0 +1,273 @@
+//! Command line parsing and execution.
+
+// The verbosity stuff is cribbed from https://github.com/clap-rs/clap-verbosity-flag/blob/c621a6a8a7c0b6df8f1464a985a5d076b4915693/src/lib.rs and updated for tracing
+
+mod item;
+mod migrate;
+mod search;
+mod serve;
+mod translate;
+mod validate;
+
+use crate::{Format, 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::metadata::Level;
+
+const BUFFER: usize = 100;
+
+/// Arguments, as parsed from the command line (usually).
+#[derive(Parser, Debug)]
+#[command(author, version, about, long_about = None)]
+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>,
+
+    /// 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>,
+
+    /// If output is being written to a file, create any directories in that file's path
+    #[arg(long, global = true)]
+    create_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>,
+
+    #[arg(
+        long,
+        short = 'v',
+        action = clap::ArgAction::Count,
+        global = true,
+        help = ErrorLevel::verbose_help(),
+        long_help = ErrorLevel::verbose_long_help(),
+    )]
+    verbose: u8,
+
+    #[arg(
+        long,
+        short = 'q',
+        action = clap::ArgAction::Count,
+        global = true,
+        help = ErrorLevel::quiet_help(),
+        long_help = ErrorLevel::quiet_long_help(),
+        conflicts_with = "verbose",
+    )]
+    quiet: u8,
+
+    /// The subcommand to run.
+    #[command(subcommand)]
+    subcommand: Subcommand,
+}
+
+/// A sucommand.
+#[derive(Debug, clap::Subcommand)]
+#[allow(clippy::large_enum_variant)]
+pub enum Subcommand {
+    /// Create a STAC Item from an id or the href to an asset
+    Item(item::Args),
+
+    /// Migrate a STAC value from one version to another
+    Migrate(migrate::Args),
+
+    /// Search for STAC items
+    Search(search::Args),
+
+    /// Serve a STAC API
+    Serve(serve::Args),
+
+    /// Translate STAC values between formats
+    Translate(translate::Args),
+
+    /// Validate a STAC object using json-schema
+    Validate(validate::Args),
+}
+
+#[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>>;
+
+    fn take_outfile(&mut self) -> Option<String> {
+        None
+    }
+}
+
+impl Args {
+    /// Returns the tracing log level for these args.
+    pub fn log_level(&self) -> Option<Level> {
+        level_enum(self.verbosity())
+    }
+
+    fn verbosity(&self) -> i8 {
+        level_value(ErrorLevel::default()) - (self.quiet as i8) + (self.verbose as i8)
+    }
+
+    /// 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)?;
+                    }
+                }
+                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)?;
+            }
+            Ok((writer, format))
+        });
+        let value = self
+            .subcommand
+            .run(
+                Input {
+                    format: self.input_format,
+                },
+                sender,
+            )
+            .await?;
+        let (writer, format) = receiver.await??;
+        if let Some(value) = value {
+            format.to_writer(writer, value)?;
+        }
+        Ok(())
+    }
+}
+
+impl Run for Subcommand {
+    async fn run(self, input: Input, sender: Sender<Value>) -> Result<Option<Value>> {
+        match self {
+            Subcommand::Item(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,
+        }
+    }
+
+    fn take_outfile(&mut self) -> Option<String> {
+        match self {
+            Subcommand::Item(args) => args.take_outfile(),
+            Subcommand::Migrate(args) => args.take_outfile(),
+            Subcommand::Search(args) => args.take_outfile(),
+            Subcommand::Serve(args) => args.take_outfile(),
+            Subcommand::Translate(args) => args.take_outfile(),
+            Subcommand::Validate(args) => args.take_outfile(),
+        }
+    }
+}
+
+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)
+    }
+
+    fn verbose_help() -> Option<&'static str> {
+        Some("Increase verbosity")
+    }
+
+    fn verbose_long_help() -> Option<&'static str> {
+        None
+    }
+
+    fn quiet_help() -> Option<&'static str> {
+        Some("Decrease verbosity")
+    }
+
+    fn quiet_long_help() -> Option<&'static str> {
+        None
+    }
+}
+
+fn level_enum(verbosity: i8) -> Option<Level> {
+    match verbosity {
+        i8::MIN..=-1 => None,
+        0 => Some(Level::ERROR),
+        1 => Some(Level::WARN),
+        2 => Some(Level::INFO),
+        3 => Some(Level::DEBUG),
+        4..=i8::MAX => Some(Level::TRACE),
+    }
+}
+
+fn level_value(level: Option<Level>) -> i8 {
+    match level {
+        None => -1,
+        Some(Level::ERROR) => 0,
+        Some(Level::WARN) => 1,
+        Some(Level::INFO) => 2,
+        Some(Level::DEBUG) => 3,
+        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
new file mode 100644
index 00000000..6b40a97f
--- /dev/null
+++ b/cli/src/args/search.rs
@@ -0,0 +1,230 @@
+use super::{Input, Run};
+use crate::{Error, Result, Value};
+use serde::de::DeserializeOwned;
+use stac_api::{GetItems, ItemCollection, Items, Search};
+use stac_async::ApiClient;
+use std::{fs::File, io::BufReader};
+use tokio::sync::mpsc::Sender;
+use tokio_stream::StreamExt;
+use tracing::info;
+
+#[derive(Debug, clap::Args)]
+pub struct Args {
+    /// The href of the STAC API or the stac-geoparquet file to search
+    #[cfg(feature = "duckdb")]
+    href: String,
+
+    /// The href of the STAC API to search
+    #[cfg(not(feature = "duckdb"))]
+    href: String,
+
+    /// The maximum number of items to return
+    #[arg(short, long)]
+    max_items: Option<usize>,
+
+    /// The maximum number of results to return from the server per page.
+    #[arg(short, long)]
+    limit: Option<u64>,
+
+    /// Requested bounding box.
+    #[arg(short, long)]
+    bbox: Option<String>,
+
+    /// Single date+time, or a range ('/' separator), formatted to RFC 3339, section 5.6. Use double dots .. for open date ranges.
+    #[arg(short, long)]
+    datetime: Option<String>,
+
+    /// Searches items by performing intersection between their geometry and
+    /// provided GeoJSON geometry. If prefixed with a `@`, the data will be read from a the file.
+    #[arg(long)]
+    intersects: Option<String>,
+
+    /// Comma-delimited list of one ore more item ids to return
+    #[arg(long, num_args=0.., value_delimiter=',')]
+    ids: Vec<String>,
+
+    /// Comma-delimited list of one or more collection IDs that each matching item must be in.
+    #[arg(short, long, num_args=0.., value_delimiter=',')]
+    collections: Vec<String>,
+
+    /// Include/exclude fields from item collections, e.g. `id,geometry,properties.datetime``
+    #[arg(long)]
+    fields: Option<String>,
+
+    /// Fields by which to sort results, e.g. `+properties.created`
+    #[arg(long)]
+    sortby: Option<String>,
+
+    /// Recommended to not be passed, but server must only accept
+    /// <http://www.opengis.net/def/crs/OGC/1.3/CRS84> as a valid value, may
+    /// reject any others
+    #[arg(long)]
+    filter_crs: Option<String>,
+
+    /// `cql2-text` or `cql2-json`. If undefined, defaults to cql2-text for a GET request and cql2-json for a POST request.
+    #[arg(long)]
+    filter_lang: Option<String>,
+
+    /// CQL2 filter expression.
+    #[arg(short, long)]
+    filter: Option<String>,
+
+    /// A query to pass, if prefixed with a `@` will be read from a file.
+    #[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.
+    ///
+    /// If true, or if not provided and the href's extension is `parquet` or `geoparquet`, DuckDB will be used to search. Set to `false` to always use a STAC API client.
+    #[arg(long)]
+    #[cfg(feature = "duckdb")]
+    duckdb: Option<bool>,
+
+    /// The file to write the output to.
+    ///
+    /// If not provided, the output will be written to standard output.
+    outfile: Option<String>,
+}
+
+async fn search_api(
+    href: String,
+    search: Search,
+    sender: Option<Sender<Value>>,
+    max_items: Option<usize>,
+) -> Result<Option<Value>> {
+    info!("searching '{}' using an api client", href);
+    let client = ApiClient::new(&href)?;
+    let stream = client.search(search).await?;
+    tokio::pin!(stream);
+    let mut count = 0;
+    if let Some(sender) = sender {
+        while let Some(item) = stream.next().await {
+            let item = item?;
+            count += 1;
+            sender.send(item.into()).await?;
+            if max_items
+                .map(|max_items| count >= max_items)
+                .unwrap_or_default()
+            {
+                break;
+            }
+        }
+        Ok(None)
+    } else {
+        let mut items = if let Some(max_items) = max_items {
+            Vec::with_capacity(max_items)
+        } else {
+            Vec::new()
+        };
+        while let Some(item) = stream.next().await {
+            items.push(item?);
+            count += 1;
+            if max_items
+                .map(|max_items| count >= max_items)
+                .unwrap_or_default()
+            {
+                break;
+            }
+        }
+        let item_collection = ItemCollection::new(items)?;
+        Ok(Some(Value::Json(serde_json::to_value(item_collection)?)))
+    }
+}
+
+#[cfg(feature = "duckdb")]
+async fn search_geoparquet(
+    href: String,
+    mut search: Search,
+    sender: Option<Sender<Value>>,
+    max_items: Option<usize>,
+) -> Result<Option<Value>> {
+    info!("searching '{}' using duckdb", href);
+    if let Some(max_items) = max_items {
+        let max_items: u64 = max_items.try_into()?;
+        if search.limit.map(|limit| limit > max_items).unwrap_or(true) {
+            search.limit = Some(max_items);
+        }
+    }
+    let client = stac_duckdb::Client::from_href(href)?;
+    let items = client.search_to_json(search)?;
+    if let Some(sender) = sender {
+        for item in items.items {
+            sender.send(item.into()).await?;
+        }
+        Ok(None)
+    } else {
+        Ok(Some(Value::Json(serde_json::to_value(items)?)))
+    }
+}
+
+impl Run for Args {
+    async fn run(self, _: Input, sender: Sender<Value>) -> Result<Option<Value>> {
+        let get_items = GetItems {
+            bbox: self.bbox,
+            datetime: self.datetime,
+            fields: self.fields,
+            sortby: self.sortby,
+            filter_crs: self.filter_crs,
+            filter_lang: self.filter_lang,
+            filter: self.filter,
+            ..Default::default()
+        };
+        let mut items: Items = get_items.try_into()?;
+        items.limit = self.limit;
+        items.query = self.query.map(json).transpose()?;
+        let mut search: Search = items.into();
+        search.intersects = self.intersects.map(json).transpose()?;
+        search.ids = if self.ids.is_empty() {
+            None
+        } else {
+            Some(self.ids)
+        };
+        search.collections = if self.collections.is_empty() {
+            None
+        } 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
+            } else {
+                search_api(self.href, search, sender, 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, sender, self.max_items).await
+        }
+    }
+
+    fn take_outfile(&mut self) -> Option<String> {
+        self.outfile.take()
+    }
+}
+
+fn json<T>(s: String) -> Result<T>
+where
+    T: DeserializeOwned,
+{
+    if s.starts_with('@') && s.len() > 1 {
+        info!("reading {}", &s[1..]);
+        let file = BufReader::new(File::open(&s[1..])?);
+        serde_json::from_reader(file).map_err(Error::from)
+    } else {
+        serde_json::from_str(&s).map_err(Error::from)
+    }
+}
diff --git a/cli/src/args/serve.rs b/cli/src/args/serve.rs
new file mode 100644
index 00000000..555ed54f
--- /dev/null
+++ b/cli/src/args/serve.rs
@@ -0,0 +1,249 @@
+use super::{Input, Run};
+use crate::{Error, Result, Value};
+use stac::{Collection, Href, Item, Links};
+use stac_server::{Api, Backend as _, MemoryBackend};
+use std::collections::{HashMap, HashSet};
+use tokio::{net::TcpListener, sync::mpsc::Sender, task::JoinSet};
+use tracing::{info, warn};
+
+const DEFAULT_COLLECTION_ID: &str = "auto-generated-collection";
+
+/// Arguments for serving an API.
+#[derive(Debug, clap::Args)]
+pub 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.
+    hrefs: Vec<String>,
+
+    /// The address of the server.
+    #[arg(short, long, default_value = "127.0.0.1:7822")]
+    addr: String,
+
+    /// The pgstac connection string, e.g. `postgresql://username:password@localhost:5432/postgis`
+    ///
+    /// If not provided an in-process memory backend will be used.
+    #[arg(long)]
+    #[cfg(feature = "pgstac")]
+    pgstac: Option<String>,
+
+    /// After loading a collection, load all of its item links
+    #[arg(long, default_value_t = true)]
+    load_collection_items: bool,
+
+    /// Create collections for any items that don't have one
+    #[arg(long, default_value_t = true)]
+    create_collections: bool,
+}
+
+#[derive(Debug, Default)]
+struct Info {
+    backend: &'static str,
+    collections: usize,
+    items: usize,
+}
+
+impl Run for Args {
+    async fn run(self, input: Input, sender: Sender<Value>) -> Result<Option<Value>> {
+        use stac::Value;
+
+        let mut info = Info::default();
+        let mut backend = {
+            #[cfg(feature = "pgstac")]
+            if let Some(pgstac) = self.pgstac {
+                info!("creating a pgstac backend at {}", pgstac);
+                info.backend = "pgstac";
+                Backend::Pgstac(stac_server::PgstacBackend::new_from_stringlike(pgstac).await?)
+            } else {
+                info!("using a memory backend");
+                info.backend = "memory";
+                Backend::Memory(MemoryBackend::new())
+            }
+            #[cfg(not(feature = "pgstac"))]
+            {
+                info!("using a memory backend");
+                info.backend = "memory";
+                Backend::Memory(MemoryBackend::new())
+            }
+        };
+
+        // 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 reading_from_stdin = false;
+        for href in self.hrefs {
+            if href == "-" {
+                if reading_from_stdin {
+                    continue;
+                } else {
+                    reading_from_stdin = true;
+                }
+            }
+            let input = input.clone();
+            let _ = join_set.spawn(async move { input.read(Some(href)) });
+        }
+        let mut item_join_set = JoinSet::new();
+        let mut collections = HashSet::new();
+        let mut items: HashMap<Option<String>, Vec<Item>> = HashMap::new();
+        while let Some(result) = join_set.join_next().await {
+            match result?? {
+                Value::Catalog(catalog) => {
+                    warn!(
+                        "cannot load catalog with id '{}' into the server",
+                        catalog.id
+                    );
+                }
+                Value::Collection(mut collection) => {
+                    if self.load_collection_items {
+                        let href = collection.href().expect("we just read it").to_string();
+                        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 _ = collections.insert(collection.id.clone());
+                    backend.add_collection(collection).await?;
+                    info.collections += 1;
+                }
+                Value::Item(item) => items.entry(item.collection.clone()).or_default().push(item),
+                Value::ItemCollection(item_collection) => {
+                    for item in item_collection {
+                        items.entry(item.collection.clone()).or_default().push(item);
+                    }
+                }
+            }
+        }
+        while let Some(result) = item_join_set.join_next().await {
+            let value = result??;
+            if let Value::Item(item) = value {
+                items.entry(item.collection.clone()).or_default().push(item);
+            } else {
+                warn!("discarding non-item of type '{}'", value.type_name());
+            }
+        }
+        if collections.contains(DEFAULT_COLLECTION_ID) && items.contains_key(&None) {
+            warn!("a collection already exists with the default collection id '{}', discarding all items without a collection", DEFAULT_COLLECTION_ID);
+            let _ = items.remove(&None);
+        }
+        for collection in collections {
+            if let Some(items) = items.remove(&Some(collection)) {
+                info.items += items.len();
+                backend.add_items(items).await?;
+            }
+        }
+        if self.create_collections {
+            for (collection, items) in items
+                .into_iter()
+                .map(|(c, i)| (c.unwrap_or(DEFAULT_COLLECTION_ID.to_string()), i))
+            {
+                let collection = Collection::from_id_and_items(collection, &items);
+                info.collections += 1;
+                backend.add_collection(collection).await?;
+                info.items += items.len();
+                backend.add_items(items).await?;
+            }
+        } else if !items.is_empty() {
+            let collection_ids: Vec<_> = items.into_keys().flatten().collect();
+            warn!("--create-collections=false, but some items don't have collections and will not be loaded (collection ids: {})", collection_ids.join(","));
+        }
+
+        info!("starting server");
+        match backend {
+            Backend::Memory(backend) => start_server(backend, self.addr, sender, info)
+                .await
+                .and(Ok(None)),
+            #[cfg(feature = "pgstac")]
+            Backend::Pgstac(backend) => start_server(backend, self.addr, sender, info)
+                .await
+                .and(Ok(None)),
+        }
+    }
+}
+
+async fn start_server<B>(backend: B, addr: String, sender: Sender<Value>, info: Info) -> Result<()>
+where
+    B: stac_server::Backend,
+{
+    let root = format!("http://{}", addr);
+    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?;
+    axum::serve(listener, router).await.map_err(Error::from)
+}
+
+enum Backend {
+    Memory(MemoryBackend),
+    #[cfg(feature = "pgstac")]
+    Pgstac(stac_server::PgstacBackend),
+}
+
+impl Backend {
+    async fn add_collection(&mut self, mut collection: Collection) -> Result<()> {
+        info!("adding collection with id '{}'", collection.id);
+        collection.remove_structural_links(); // TODO allow use to drain the child links first
+        match self {
+            Backend::Memory(backend) => backend
+                .add_collection(collection)
+                .await
+                .map_err(Error::from),
+            #[cfg(feature = "pgstac")]
+            Backend::Pgstac(backend) => backend
+                .add_collection(collection)
+                .await
+                .map_err(Error::from),
+        }
+    }
+
+    async fn add_items(&mut self, items: Vec<Item>) -> Result<()> {
+        assert!(!items.is_empty());
+        info!(
+            "adding {} items to collection '{}'",
+            items.len(),
+            items[0]
+                .collection
+                .as_ref()
+                .expect("all items should have a collection"),
+        );
+        for mut item in items {
+            item.remove_structural_links();
+            match self {
+                Backend::Memory(backend) => backend.add_item(item).await?,
+                #[cfg(feature = "pgstac")]
+                Backend::Pgstac(backend) => backend.add_item(item).await?,
+            }
+        }
+        Ok(())
+    }
+}
+
+impl Info {
+    fn counts(&self) -> String {
+        if self.collections > 0 || self.items > 0 {
+            let collection = if self.collections == 1 {
+                "collection"
+            } else {
+                "collections"
+            };
+            let item = if self.items == 1 { "item" } else { "items" };
+            format!(
+                " (loaded {} {} and {} {})",
+                self.collections, collection, self.items, item
+            )
+        } else {
+            String::new()
+        }
+    }
+}
diff --git a/cli/src/args/translate.rs b/cli/src/args/translate.rs
new file mode 100644
index 00000000..f5c46f88
--- /dev/null
+++ b/cli/src/args/translate.rs
@@ -0,0 +1,23 @@
+use super::{Input, Run};
+use crate::{Result, Value};
+use tokio::sync::mpsc::Sender;
+
+/// Arguments for the `translate` subcommand.
+#[derive(clap::Args, Debug)]
+pub struct Args {
+    /// The input file, if not provided or `-` the input will be read from standard input
+    infile: Option<String>,
+
+    /// The output file, if not provided the item will be printed to standard output
+    outfile: Option<String>,
+}
+
+impl Run for Args {
+    async fn run(self, input: Input, _: Sender<Value>) -> Result<Option<Value>> {
+        input.read(self.infile).map(|value| Some(value.into()))
+    }
+
+    fn take_outfile(&mut self) -> Option<String> {
+        self.outfile.take()
+    }
+}
diff --git a/cli/src/args/validate.rs b/cli/src/args/validate.rs
new file mode 100644
index 00000000..9132d484
--- /dev/null
+++ b/cli/src/args/validate.rs
@@ -0,0 +1,41 @@
+use super::{Input, Run};
+use crate::{Error, Result, Value};
+use stac_validate::Validate;
+use tokio::sync::mpsc::Sender;
+
+/// Arguments for the `validate` subcommand.
+#[derive(clap::Args, Debug)]
+pub struct Args {
+    /// The input file, if not provided or `-` the input will be read from standard input
+    infile: Option<String>,
+
+    /// The output file, if not provided the item will be printed to standard output
+    outfile: Option<String>,
+}
+
+impl Run for Args {
+    async fn run(self, input: Input, sender: Sender<Value>) -> Result<Option<Value>> {
+        let value = input.read(self.infile)?;
+        let result = value.validate();
+        if let Err(stac_validate::Error::Validation(ref errors)) = result {
+            let message_base = match value {
+                stac::Value::Item(item) => format!("[item={}] ", item.id),
+                stac::Value::Catalog(catalog) => format!("[catalog={}] ", catalog.id),
+                stac::Value::Collection(collection) => format!("[collection={}] ", collection.id),
+                stac::Value::ItemCollection(_) => "[item-collection] ".to_string(),
+            };
+            for error in errors {
+                let message = format!(
+                    "{}{} (instance path: '{}', schema path: '{}')",
+                    message_base, error, error.instance_path, error.schema_path
+                );
+                sender.send(message.into()).await?;
+            }
+        }
+        result.and(Ok(None)).map_err(Error::from)
+    }
+
+    fn take_outfile(&mut self) -> Option<String> {
+        self.outfile.take()
+    }
+}
diff --git a/cli/src/error.rs b/cli/src/error.rs
index d6f13752..1b13c534 100644
--- a/cli/src/error.rs
+++ b/cli/src/error.rs
@@ -1,14 +1,10 @@
-use crate::Output;
+use crate::Value;
 use thiserror::Error;
 
 /// Crate specific error type.
-#[derive(Error, Debug)]
+#[derive(Debug, Error)]
 #[non_exhaustive]
 pub enum Error {
-    /// Custom error.
-    #[error("{0}")]
-    Custom(String),
-
     /// [std::io::Error]
     #[error(transparent)]
     Io(#[from] std::io::Error),
@@ -48,19 +44,19 @@ pub enum Error {
 
     /// [tokio::sync::mpsc::error::SendError]
     #[error(transparent)]
-    SendOutput(#[from] tokio::sync::mpsc::error::SendError<Output>),
+    TokioSend(#[from] tokio::sync::mpsc::error::SendError<Value>),
 
     /// [tokio::task::JoinError]
     #[error(transparent)]
     TokioJoinError(#[from] tokio::task::JoinError),
 
-    /// Unsupported output format.
-    #[error("unsupported output format: {0}")]
-    UnsupportedFormat(String),
+    /// [std::num::TryFromIntError]
+    #[error(transparent)]
+    TryFromInt(#[from] std::num::TryFromIntError),
 
-    /// Validation errors.
-    #[error("validation errors: {0:?}")]
-    Validation(Vec<serde_json::Value>),
+    /// Unsupported format.
+    #[error("unsupported (or unknown) format: {0}")]
+    UnsupportedFormat(String),
 }
 
 impl Error {
diff --git a/cli/src/format.rs b/cli/src/format.rs
index 6b530ebf..63817927 100644
--- a/cli/src/format.rs
+++ b/cli/src/format.rs
@@ -1,53 +1,201 @@
-use crate::{Error, Result};
-use clap::ValueEnum;
-use serde::de::DeserializeOwned;
+use crate::{Error, Result, Value};
+use stac::{Item, ItemCollection};
+use std::{
+    io::{BufRead, BufReader, Read, Write},
+    str::FromStr,
+};
 
-/// The STAC output format.
-#[derive(Clone, Copy, Debug, Default, ValueEnum)]
+/// A file format that we can read and/or write.
+#[derive(Clone, Debug, PartialEq)]
 pub enum Format {
-    #[cfg(feature = "geoparquet")]
+    /// Pretty-printed JSON
+    PrettyJson,
+
+    /// Compact-printed JSON
+    CompactJson,
+
+    /// Streaming data.
+    ///
+    /// For JSON, this is newline-delimited JSON.
+    Streaming,
+
     /// stac-geoparquet
-    Parquet,
+    #[cfg(feature = "geoparquet")]
+    Geoparquet(Option<parquet::basic::Compression>),
+}
 
-    /// JSON (the default).
-    #[default]
-    Json,
+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 {
-    pub(crate) fn maybe_from_href(href: &str) -> Option<Format> {
-        href.rsplit_once('.')
-            .and_then(|(_, ext)| Format::from_str(ext, true).ok())
+    /// 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
+        })
     }
 
-    pub(crate) async fn read_href<D: DeserializeOwned>(&self, href: Option<&str>) -> Result<D> {
-        if let Some(href) = href.and_then(|href| if href == "-" { None } else { Some(href) }) {
-            match *self {
-                #[cfg(feature = "geoparquet")]
-                Format::Parquet => {
-                    let item_collection = if let Some(url) = stac::href_to_url(href) {
-                        stac::geoparquet::from_reader(reqwest::blocking::get(url)?.bytes()?)?
-                    } else {
-                        let file = std::fs::File::open(href)?;
-                        stac::geoparquet::from_reader(file)?
-                    };
-                    serde_json::from_value(serde_json::to_value(item_collection)?)
-                        .map_err(Error::from)
+    /// 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),
                 }
-                Format::Json => stac_async::read_json(href).await.map_err(Error::from),
             }
-        } else {
-            match *self {
-                #[cfg(feature = "geoparquet")]
-                Format::Parquet => {
-                    use std::io::Read;
-                    let mut buf = Vec::new();
-                    let _ = std::io::stdin().read_to_end(&mut buf)?;
-                    let item_collection = stac::geoparquet::from_reader(bytes::Bytes::from(buf))?;
-                    serde_json::from_value(serde_json::to_value(item_collection)?)
-                        .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 | Format::Streaming => {
+                stac::io::json::read(file).map_err(Error::from)
+            }
+            #[cfg(feature = "geoparquet")]
+            Format::Geoparquet(_) => stac::io::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);
                 }
-                Format::Json => serde_json::from_reader(std::io::stdin()).map_err(Error::from),
+                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/lib.rs b/cli/src/lib.rs
index eedfae54..bb4720a0 100644
--- a/cli/src/lib.rs
+++ b/cli/src/lib.rs
@@ -31,112 +31,19 @@
     warnings
 )]
 
-mod args;
+pub mod args;
 mod error;
 mod format;
-mod output;
-mod runner;
-mod subcommand;
+#[cfg(feature = "python")]
+mod python;
+mod value;
 
-pub use {
-    args::{
-        Args, ItemArgs, MigrateArgs, SearchArgs, ServeArgs, SortArgs, TranslateArgs, ValidateArgs,
-    },
-    error::Error,
-    format::Format,
-    output::Output,
-    runner::Runner,
-    subcommand::Subcommand,
-};
+pub use {args::Args, error::Error, format::Format, value::Value};
 
 /// Crate-specific result type.
 pub type Result<T> = std::result::Result<T, Error>;
 
-/// Run the command-line interface.
-///
-/// # Examples
-///
-/// ```
-/// use stac_cli::{Args, Subcommand, SortArgs};
-///
-/// let sort_args = SortArgs {
-///     infile: Some("examples/simple-item.json".to_string()),
-///     outfile: None,
-/// };
-/// let args = Args {
-///     compact: false,
-///     input_format: None,
-///     output_format: None,
-///     #[cfg(feature = "geoparquet")]
-///     geoparquet_compression: None,
-///     subcommand: Subcommand::Sort(sort_args),
-/// };
-/// # tokio_test::block_on(async {
-/// stac_cli::run(args).await.unwrap();
-/// # })
-/// ```
-pub async fn run(args: Args) -> Result<()> {
-    let writer = args.writer()?;
-    let outfile = args.outfile().map(String::from);
-    let input_format = args.input_format();
-    let output_format = args.output_format();
-    let mut runner = Runner {
-        compact: args.compact,
-        input_format,
-        output_format,
-        #[cfg(feature = "geoparquet")]
-        geoparquet_compression: args.geoparquet_compression,
-        writer,
-        buffer: 100,
-    };
-    let result = runner.run(args.subcommand).await;
-    if result.is_err() {
-        if let Some(outfile) = outfile {
-            if let Err(err) = std::fs::remove_file(outfile) {
-                eprintln!("error when unlinking outfile: {}", err);
-            }
-        }
-    }
-    result
-}
-
-#[cfg(feature = "python")]
-mod python {
-    use crate::Args;
-    use clap::Parser;
-    use pyo3::{
-        prelude::{PyModule, PyModuleMethods},
-        pyfunction, pymodule, wrap_pyfunction, Bound, PyResult,
-    };
-
-    #[pyfunction]
-    fn main() -> PyResult<i64> {
-        std::process::exit(
-            tokio::runtime::Builder::new_multi_thread()
-                .enable_all()
-                .build()
-                .unwrap()
-                .block_on(async {
-                    // We skip one because the first argument is going to be the python interpreter.
-                    let args = Args::parse_from(std::env::args_os().skip(1));
-                    match super::run(args).await {
-                        Ok(()) => 0,
-                        Err(err) => {
-                            eprintln!("ERROR: {}", err);
-                            err.code()
-                        }
-                    }
-                }),
-        )
-    }
-
-    #[pymodule]
-    fn stacrs_cli(m: &Bound<'_, PyModule>) -> PyResult<()> {
-        m.add_function(wrap_pyfunction!(main, m)?)?;
-        Ok(())
-    }
-}
-
 use tracing_subscriber as _;
+
 #[cfg(test)]
 use {assert_cmd as _, tokio_test as _};
diff --git a/cli/src/main.rs b/cli/src/main.rs
index 34870a5f..100391b9 100644
--- a/cli/src/main.rs
+++ b/cli/src/main.rs
@@ -1,15 +1,13 @@
 use clap::Parser;
 use stac_cli::Args;
-use tracing_subscriber;
 
 #[tokio::main]
 async fn main() {
-    if std::env::var("STACRS_TRACING").is_ok() {
-        tracing_subscriber::fmt::init();
-    }
-
     let args = Args::parse();
-    std::process::exit(match stac_cli::run(args).await {
+    tracing_subscriber::fmt()
+        .with_max_level(args.log_level())
+        .init();
+    std::process::exit(match args.run(std::io::stdout()).await {
         Ok(()) => 0,
         Err(err) => {
             eprintln!("ERROR: {}", err);
diff --git a/cli/src/output.rs b/cli/src/output.rs
deleted file mode 100644
index 6f9ea496..00000000
--- a/cli/src/output.rs
+++ /dev/null
@@ -1,86 +0,0 @@
-use std::fmt::Display;
-
-/// CLI output.
-#[derive(Debug)]
-pub enum Output {
-    /// A STAC value.
-    Stac(stac::Value),
-
-    /// A serde_json value.
-    Json(serde_json::Value),
-
-    /// A string.
-    String(String),
-}
-
-impl Output {
-    /// Converts this output to Json.
-    ///
-    /// Strings are not converted.
-    pub fn to_json(&self) -> Option<serde_json::Value> {
-        match self {
-            Output::Stac(value) => serde_json::to_value(value).ok(),
-            Output::Json(value) => Some(value.clone()),
-            Output::String(_) => None,
-        }
-    }
-
-    /// Converts this output to [stac::Value].
-    ///
-    /// Strings are not converted.
-    pub fn to_stac(&self) -> Option<stac::Value> {
-        match self {
-            Output::Stac(value) => Some(value.clone()),
-            Output::Json(value) => serde_json::from_value(value.clone()).ok(),
-            Output::String(_) => None,
-        }
-    }
-}
-
-impl From<stac::Value> for Output {
-    fn from(value: stac::Value) -> Self {
-        Output::Stac(value)
-    }
-}
-
-impl From<serde_json::Value> for Output {
-    fn from(value: serde_json::Value) -> Self {
-        Output::Json(value)
-    }
-}
-
-impl From<serde_json::Map<String, serde_json::Value>> for Output {
-    fn from(value: serde_json::Map<String, serde_json::Value>) -> Self {
-        Output::Json(serde_json::Value::Object(value))
-    }
-}
-
-impl From<String> for Output {
-    fn from(value: String) -> Self {
-        Output::String(value)
-    }
-}
-
-impl From<&str> for Output {
-    fn from(value: &str) -> Self {
-        Output::String(value.to_string())
-    }
-}
-
-impl Display for Output {
-    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
-        match self {
-            Output::Stac(value) => write!(
-                f,
-                "{}",
-                serde_json::to_string(value).unwrap_or_else(|err| err.to_string())
-            ),
-            Output::Json(value) => write!(
-                f,
-                "{}",
-                serde_json::to_string(value).unwrap_or_else(|err| err.to_string())
-            ),
-            Output::String(string) => write!(f, "{}", string),
-        }
-    }
-}
diff --git a/cli/src/python.rs b/cli/src/python.rs
new file mode 100644
index 00000000..e39bf4ac
--- /dev/null
+++ b/cli/src/python.rs
@@ -0,0 +1,36 @@
+use crate::Args;
+use clap::Parser;
+use pyo3::{
+    prelude::{PyModule, PyModuleMethods},
+    pyfunction, pymodule, wrap_pyfunction, Bound, PyResult,
+};
+
+#[pyfunction]
+fn main() -> PyResult<i64> {
+    let args = Args::parse_from(std::env::args_os().skip(1));
+    tracing_subscriber::fmt()
+        .with_max_level(args.log_level())
+        .init();
+    std::process::exit(
+        tokio::runtime::Builder::new_multi_thread()
+            .enable_all()
+            .build()
+            .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 {
+                    Ok(()) => 0,
+                    Err(err) => {
+                        eprintln!("ERROR: {}", err);
+                        err.code()
+                    }
+                }
+            }),
+    )
+}
+
+#[pymodule]
+fn stacrs_cli(m: &Bound<'_, PyModule>) -> PyResult<()> {
+    m.add_function(wrap_pyfunction!(main, m)?)?;
+    Ok(())
+}
diff --git a/cli/src/runner.rs b/cli/src/runner.rs
deleted file mode 100644
index 23605d3f..00000000
--- a/cli/src/runner.rs
+++ /dev/null
@@ -1,77 +0,0 @@
-use crate::{Format, Result, Subcommand};
-use std::io::Write;
-
-/// Struct for running commands.
-#[derive(Debug)]
-pub struct Runner<W: Write>
-where
-    W: Send,
-{
-    /// Should the output be printed in compact form, if supported?
-    pub compact: bool,
-
-    /// The input format.
-    pub input_format: Format,
-
-    /// The output format.
-    pub output_format: Format,
-
-    /// Geoparquet compression.
-    #[cfg(feature = "geoparquet")]
-    pub geoparquet_compression: Option<parquet::basic::Compression>,
-
-    /// The output writeable stream.
-    pub writer: W,
-
-    /// The size of the message channel buffer.
-    pub buffer: usize,
-}
-
-impl<W: Write> Runner<W>
-where
-    W: Send,
-{
-    pub(crate) async fn run(&mut self, subcommand: Subcommand) -> Result<()> {
-        let (sender, mut receiver) = tokio::sync::mpsc::channel(self.buffer);
-        let input_format = self.input_format;
-        let handle = tokio::spawn(async move { subcommand.run(input_format, sender).await });
-        while let Some(value) = receiver.recv().await {
-            match self.output_format {
-                Format::Json => {
-                    if let Some(value) = value.to_json() {
-                        if self.compact {
-                            serde_json::to_writer(&mut self.writer, &value)?;
-                            writeln!(&mut self.writer)?;
-                        } else {
-                            serde_json::to_writer_pretty(&mut self.writer, &value)?;
-                            writeln!(&mut self.writer)?;
-                        }
-                    } else {
-                        writeln!(self.writer, "{}", value)?;
-                    }
-                }
-                #[cfg(feature = "geoparquet")]
-                Format::Parquet => {
-                    if let Some(value) = value.to_stac() {
-                        let mut options = geoarrow::io::parquet::GeoParquetWriterOptions::default();
-                        if let Some(compression) = self.geoparquet_compression {
-                            let writer_properites =
-                                parquet::file::properties::WriterProperties::builder()
-                                    .set_compression(compression)
-                                    .build();
-                            options.writer_properties = Some(writer_properites);
-                        }
-                        stac::geoparquet::to_writer_with_options(
-                            &mut self.writer,
-                            value,
-                            &options,
-                        )?;
-                    } else {
-                        writeln!(self.writer, "{}", value)?;
-                    }
-                }
-            }
-        }
-        handle.await?
-    }
-}
diff --git a/cli/src/subcommand/item.rs b/cli/src/subcommand/item.rs
deleted file mode 100644
index 39b9a7a4..00000000
--- a/cli/src/subcommand/item.rs
+++ /dev/null
@@ -1,70 +0,0 @@
-use crate::{Error, ItemArgs, Result, Subcommand};
-use stac::{item::Builder, Asset, Value};
-use std::path::Path;
-use url::Url;
-
-impl Subcommand {
-    pub(crate) fn item(args: ItemArgs) -> Result<Value> {
-        let mut disable_gdal = args.disable_gdal;
-        if !(disable_gdal || cfg!(feature = "gdal")) {
-            tracing::info!(disable_gdal = true, "gdal feature not enabled");
-            disable_gdal = true;
-        }
-        let mut href = None;
-        let id = if let Ok(url) = Url::parse(&args.id_or_href) {
-            href = Some(args.id_or_href.clone());
-            args.id
-                .as_ref()
-                .map(|id| id.to_string())
-                .unwrap_or_else(|| {
-                    url.path_segments()
-                        .and_then(|path_segments| path_segments.last())
-                        .and_then(|path_segment| Path::new(path_segment).file_stem())
-                        .map(|file_stem| file_stem.to_string_lossy().into_owned())
-                        .unwrap_or_else(|| args.id_or_href.to_string())
-                })
-        } else {
-            let path = Path::new(&args.id_or_href);
-            if path.exists() {
-                href = Some(args.id_or_href.clone());
-                args.id
-                    .as_ref()
-                    .map(|id| id.to_string())
-                    .unwrap_or_else(|| {
-                        path.file_stem()
-                            .map(|file_stem| file_stem.to_string_lossy().into_owned())
-                            .unwrap_or_else(|| args.id_or_href.to_string())
-                    })
-            } else {
-                args.id_or_href.to_string()
-            }
-        };
-        let mut builder = Builder::new(id)
-            .enable_gdal(!disable_gdal)
-            .canonicalize_paths(!args.allow_relative_paths);
-        if let Some(href) = href {
-            let mut asset = Asset::new(href);
-            asset.roles = args.role.to_vec();
-            builder = builder.asset(&args.key, asset);
-        }
-        let item = builder.into_item()?;
-        if args.collect {
-            let value = serde_json::from_reader(std::io::stdin())?;
-            match value {
-                Value::Item(stdin_item) => {
-                    Ok(stac::ItemCollection::from(vec![stdin_item, item]).into())
-                }
-                Value::ItemCollection(mut item_collection) => {
-                    item_collection.items.push(item);
-                    Ok(item_collection.into())
-                }
-                Value::Catalog(_) | Value::Collection(_) => Err(Error::Custom(format!(
-                    "unexpected STAC object type on standard input: {}",
-                    value.type_name()
-                ))),
-            }
-        } else {
-            Ok(item.into())
-        }
-    }
-}
diff --git a/cli/src/subcommand/migrate.rs b/cli/src/subcommand/migrate.rs
deleted file mode 100644
index 5e8a0734..00000000
--- a/cli/src/subcommand/migrate.rs
+++ /dev/null
@@ -1,10 +0,0 @@
-use crate::{Error, Format, MigrateArgs, Result, Subcommand};
-use stac::{Migrate, Value};
-
-impl Subcommand {
-    /// Migrates a STAC value.
-    pub(crate) async fn migrate(args: MigrateArgs, input_format: Format) -> Result<Value> {
-        let value: Value = input_format.read_href(args.infile.as_deref()).await?;
-        value.migrate(args.version).map_err(Error::from)
-    }
-}
diff --git a/cli/src/subcommand/mod.rs b/cli/src/subcommand/mod.rs
deleted file mode 100644
index ca12dcbe..00000000
--- a/cli/src/subcommand/mod.rs
+++ /dev/null
@@ -1,110 +0,0 @@
-mod item;
-mod migrate;
-mod search;
-mod serve;
-mod sort;
-mod translate;
-mod validate;
-
-use crate::{
-    Error, Format, ItemArgs, MigrateArgs, Output, Result, SearchArgs, ServeArgs, SortArgs,
-    TranslateArgs, ValidateArgs,
-};
-use tokio::sync::mpsc::Sender;
-
-/// A CLI subcommand.
-#[derive(Debug, clap::Subcommand)]
-#[allow(clippy::large_enum_variant)]
-pub enum Subcommand {
-    /// Creates a STAC Item.
-    Item(ItemArgs),
-
-    /// Migrates a STAC value from one version to another.
-    Migrate(MigrateArgs),
-
-    /// Searches a STAC API.
-    Search(SearchArgs),
-
-    /// Serves a STAC API.
-    ///
-    /// By default, uses a basic memory backend, which is not suitable for
-    /// production. To use the pgstac backend, pass the pgstac connection string
-    /// to the `--pgstac` argument.
-    Serve(ServeArgs),
-
-    /// Sorts the fields of STAC object.
-    Sort(SortArgs),
-
-    /// Translates STAC values between formats.
-    Translate(TranslateArgs),
-
-    /// Validates a STAC object or API endpoint using json-schema validation.
-    Validate(ValidateArgs),
-}
-
-impl Subcommand {
-    pub(crate) fn infile(&self) -> Option<&str> {
-        use Subcommand::*;
-
-        match self {
-            Item(args) => Some(args.id_or_href.as_str()),
-            Sort(args) => args.infile.as_deref(),
-            Translate(args) => args.infile.as_deref(),
-            Validate(args) => args.href.as_deref(),
-            _ => None,
-        }
-    }
-
-    pub(crate) fn outfile(&self) -> Option<&str> {
-        use Subcommand::*;
-
-        match self {
-            Item(args) => args.outfile.as_deref(),
-            Search(args) => args.outfile.as_deref(),
-            Sort(args) => args.outfile.as_deref(),
-            Translate(args) => args.outfile.as_deref(),
-            _ => None,
-        }
-    }
-
-    pub(crate) async fn run(self, input_format: Format, sender: Sender<Output>) -> Result<()> {
-        use Subcommand::*;
-
-        match self {
-            Item(args) => {
-                let item = Subcommand::item(args)?;
-                sender.send(item.into()).await?;
-            }
-            Migrate(args) => {
-                let value = Subcommand::migrate(args, input_format).await?;
-                sender.send(value.into()).await?;
-            }
-            Search(args) => Subcommand::search(args, sender).await?,
-            Serve(args) => Subcommand::serve(args, sender).await?,
-            Sort(args) => {
-                let value = Subcommand::sort(args, input_format).await?;
-                sender.send(value.into()).await?;
-            }
-            Translate(args) => {
-                let value = Subcommand::translate(args, input_format).await?;
-                sender.send(value.into()).await?;
-            }
-            Validate(args) => {
-                if let Err(err) = Subcommand::validate(args, input_format).await {
-                    match err {
-                        Error::Validation(errors) => {
-                            for error in errors {
-                                sender.send(error.into()).await?;
-                            }
-                            sender
-                                .send("ERROR: one or more validation errors".into())
-                                .await?;
-                        }
-                        _ => return Err(err),
-                    }
-                }
-            }
-        }
-        Ok(())
-    }
-}
diff --git a/cli/src/subcommand/search.rs b/cli/src/subcommand/search.rs
deleted file mode 100644
index cf0f5c30..00000000
--- a/cli/src/subcommand/search.rs
+++ /dev/null
@@ -1,90 +0,0 @@
-use crate::{Output, Result, SearchArgs, Subcommand};
-use stac_api::{GetItems, GetSearch, Item, ItemCollection};
-use stac_async::ApiClient;
-use tokio::sync::mpsc::Sender;
-use tokio_stream::StreamExt;
-
-impl Subcommand {
-    pub(crate) async fn search(args: SearchArgs, sender: Sender<Output>) -> Result<()> {
-        let get_items = GetItems {
-            limit: args.limit,
-            bbox: args.bbox,
-            datetime: args.datetime,
-            fields: args.fields,
-            sortby: args.sortby,
-            filter_crs: args.filter_crs,
-            filter_lang: args.filter_lang,
-            filter: args.filter,
-            additional_fields: Default::default(),
-        };
-        let get_search = GetSearch {
-            intersects: args.intersects,
-            ids: args.ids,
-            collections: args.collections,
-            items: get_items,
-        };
-        let search = get_search.try_into()?;
-        if args
-            .duckdb
-            .unwrap_or_else(|| stac::geoparquet::has_extension(&args.href))
-        {
-            #[cfg(feature = "duckdb")]
-            {
-                let client = stac_duckdb::Client::from_href(args.href)?;
-                let items = client.search_to_json(search)?;
-                if args.stream {
-                    for item in items.items {
-                        sender.send(item.into()).await?;
-                    }
-                } else {
-                    sender.send(serde_json::to_value(items)?.into()).await?;
-                }
-            }
-            #[cfg(not(feature = "duckdb"))]
-            {
-                if args.duckdb == Some(true) {
-                    return Err(crate::Error::Custom(format!("`--duckdb true` was provided, but this crate was not compiled with duckdb support")));
-                } else {
-                    return Err(crate::Error::Custom(format!("The search href was auto-detected as geoparquet, but this crate was not compiled with DuckDB support. Either re-install with `-F duckdb` or disable geoparquet auto-detection with `--duckdb false`")));
-                }
-            }
-        } else {
-            let client = ApiClient::new(&args.href)?;
-            let item_stream = client.search(search).await?;
-            tokio::pin!(item_stream);
-            let mut num_items = 0;
-            if args.stream {
-                while let Some(result) = item_stream.next().await {
-                    let item: Item = result?;
-                    num_items += 1;
-                    sender.send(item.into()).await?;
-                    if args
-                        .max_items
-                        .map(|max_items| num_items >= max_items)
-                        .unwrap_or(false)
-                    {
-                        break;
-                    }
-                }
-            } else {
-                let mut items = Vec::new();
-                while let Some(result) = item_stream.next().await {
-                    num_items += 1;
-                    items.push(result?);
-                    if args
-                        .max_items
-                        .map(|max_items| num_items >= max_items)
-                        .unwrap_or(false)
-                    {
-                        break;
-                    }
-                }
-                let item_collection = ItemCollection::new(items)?;
-                sender
-                    .send(serde_json::to_value(item_collection)?.into())
-                    .await?;
-            }
-        }
-        Ok(())
-    }
-}
diff --git a/cli/src/subcommand/serve.rs b/cli/src/subcommand/serve.rs
deleted file mode 100644
index f7ca8889..00000000
--- a/cli/src/subcommand/serve.rs
+++ /dev/null
@@ -1,56 +0,0 @@
-use crate::{Output, Result, ServeArgs, Subcommand};
-use stac_server::{Api, Backend, MemoryBackend};
-use tokio::{net::TcpListener, sync::mpsc::Sender};
-
-impl Subcommand {
-    #[allow(unused_variables)]
-    pub(crate) async fn serve(args: ServeArgs, sender: Sender<Output>) -> Result<()> {
-        let root = "http://127.0.0.1:7822";
-        let addr = "127.0.0.1:7822";
-        if let Some(pgstac) = args.pgstac.as_deref() {
-            #[cfg(feature = "pgstac")]
-            {
-                let mut backend = stac_server::PgstacBackend::new_from_stringlike(pgstac).await?;
-                if !args.href.is_empty() {
-                    backend
-                        .add_from_hrefs(
-                            args.href,
-                            !args.dont_auto_create_collections,
-                            !args.dont_follow_links,
-                        )
-                        .await?;
-                }
-                let api = Api::new(backend, root)?;
-                let router = stac_server::routes::from_api(api);
-                let listener = TcpListener::bind(addr).await.unwrap();
-                sender
-                    .send(format!("Serving a STAC API at {} using a pgstac backend", root).into())
-                    .await?;
-                axum::serve(listener, router).await.unwrap();
-            }
-            #[cfg(not(feature = "pgstac"))]
-            return Err(crate::Error::Custom(
-                "stac-cli is not compiled with pgstac support".to_string(),
-            ));
-        } else {
-            let mut backend = MemoryBackend::new();
-            if !args.href.is_empty() {
-                backend
-                    .add_from_hrefs(
-                        args.href,
-                        !args.dont_auto_create_collections,
-                        !args.dont_follow_links,
-                    )
-                    .await?;
-            }
-            let api = Api::new(backend, root)?;
-            let router = stac_server::routes::from_api(api);
-            let listener = TcpListener::bind(addr).await.unwrap();
-            sender
-                .send(format!("Serving a STAC API at {} using a memory backend", root).into())
-                .await?;
-            axum::serve(listener, router).await.unwrap();
-        };
-        Ok(())
-    }
-}
diff --git a/cli/src/subcommand/sort.rs b/cli/src/subcommand/sort.rs
deleted file mode 100644
index 5b9eaafa..00000000
--- a/cli/src/subcommand/sort.rs
+++ /dev/null
@@ -1,9 +0,0 @@
-use crate::{Format, Result, SortArgs, Subcommand};
-use stac::Value;
-
-impl Subcommand {
-    /// Sorts a STAC value.
-    pub(crate) async fn sort(args: SortArgs, input_format: Format) -> Result<Value> {
-        input_format.read_href(args.infile.as_deref()).await
-    }
-}
diff --git a/cli/src/subcommand/translate.rs b/cli/src/subcommand/translate.rs
deleted file mode 100644
index d89d805f..00000000
--- a/cli/src/subcommand/translate.rs
+++ /dev/null
@@ -1,11 +0,0 @@
-use crate::{Format, Result, Subcommand, TranslateArgs};
-use stac::Value;
-use tracing::{info, instrument};
-
-impl Subcommand {
-    #[instrument]
-    pub(crate) async fn translate(args: TranslateArgs, input_format: Format) -> Result<Value> {
-        info!("reading {}", args.infile.as_deref().unwrap_or("<stdin>"));
-        input_format.read_href(args.infile.as_deref()).await
-    }
-}
diff --git a/cli/src/subcommand/validate.rs b/cli/src/subcommand/validate.rs
deleted file mode 100644
index 4516ae37..00000000
--- a/cli/src/subcommand/validate.rs
+++ /dev/null
@@ -1,55 +0,0 @@
-use crate::{Error, Format, Result, Subcommand, ValidateArgs};
-use serde_json::json;
-use stac_validate::Validate;
-
-impl Subcommand {
-    /// Validates a STAC value.
-    pub async fn validate(args: ValidateArgs, input_format: Format) -> Result<()> {
-        let value: serde_json::Value = input_format.read_href(args.href.as_deref()).await?;
-        let mut errors: Vec<serde_json::Value> = Vec::new();
-        let mut update_errors = |result: std::result::Result<(), stac_validate::Error>| match result
-        {
-            Ok(()) => {}
-            Err(err) => match err {
-                stac_validate::Error::Validation(ref e) => {
-                    errors.extend(e.iter().map(|error| {
-                        json!({
-                                "type": "validation",
-                                "instance_path": error.instance_path,
-                                "schema_path": error.schema_path,
-                                "description": error.to_string(),
-                        })
-                    }));
-                }
-                _ => errors.push(json!({
-                    "type": "other",
-                    "message": err.to_string(),
-                })),
-            },
-        };
-        if let Some(collections) = value
-            .as_object()
-            .and_then(|object| object.get("collections"))
-        {
-            if let Some(collections) = collections.as_array() {
-                for collection in collections.iter() {
-                    let collection = collection.clone();
-                    let result = tokio::task::spawn_blocking(move || collection.validate()).await?;
-                    update_errors(result);
-                }
-            } else {
-                return Err(Error::Custom(
-                    "expected the 'collections' key to be an array".to_string(),
-                ));
-            }
-        } else {
-            let result = tokio::task::spawn_blocking(move || value.validate()).await?;
-            update_errors(result);
-        }
-        if errors.is_empty() {
-            Ok(())
-        } else {
-            Err(Error::Validation(errors))
-        }
-    }
-}
diff --git a/cli/src/value.rs b/cli/src/value.rs
new file mode 100644
index 00000000..4a4a28b4
--- /dev/null
+++ b/cli/src/value.rs
@@ -0,0 +1,50 @@
+use serde::Serialize;
+
+/// An output value, which can either be a [serde_json::Value], a [stac::Value], or a [String].
+#[derive(Debug, Serialize)]
+#[serde(untagged)]
+pub enum Value {
+    /// A STAC value.
+    Stac(stac::Value),
+
+    /// A JSON value.
+    Json(serde_json::Value),
+
+    /// A string value.
+    String(String),
+}
+
+impl From<stac::Value> for Value {
+    fn from(value: stac::Value) -> Self {
+        Value::Stac(value)
+    }
+}
+
+impl From<stac_api::Item> for Value {
+    fn from(value: stac_api::Item) -> Self {
+        Self::Json(value.into())
+    }
+}
+
+impl From<serde_json::Value> for Value {
+    fn from(value: serde_json::Value) -> Self {
+        Self::Json(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> {
+        match value {
+            Value::Stac(value) => Ok(value),
+            Value::Json(value) => serde_json::from_value(value),
+            Value::String(string) => serde_json::from_str(&string),
+        }
+    }
+}
diff --git a/core/README.md b/core/README.md
index d2a79b03..eceadba2 100644
--- a/core/README.md
+++ b/core/README.md
@@ -50,7 +50,7 @@ Then, items created from rasters will include the projection and raster extensio
 #[cfg(feature = "gdal")]
 {
     use stac::{extensions::{Raster, Projection}, Extensions, item::Builder};
-    let item = Builder::new("an-id").asset("data", "assets/dataset_geo.tif").into_item().unwrap();
+    let item = Builder::new("an-id").asset("data", "assets/dataset_geo.tif").build().unwrap();
     assert!(item.has_extension::<Projection>());
     assert!(item.has_extension::<Raster>());
 }
diff --git a/core/src/gdal.rs b/core/src/gdal.rs
index acf95c53..948ffa29 100644
--- a/core/src/gdal.rs
+++ b/core/src/gdal.rs
@@ -172,7 +172,7 @@ mod tests {
     fn raster_data_type() {
         let mut item = Builder::new("an-id")
             .asset("data", "assets/dataset.tif")
-            .into_item()
+            .build()
             .unwrap();
         super::update_item(&mut item, false, true).unwrap();
         assert!(item.has_extension::<Raster>());
@@ -193,7 +193,7 @@ mod tests {
     fn raster_spatial_resolution() {
         let mut item = Builder::new("an-id")
             .asset("data", "assets/dataset_geo.tif")
-            .into_item()
+            .build()
             .unwrap();
         super::update_item(&mut item, false, true).unwrap();
         let raster: Raster = item
@@ -213,7 +213,7 @@ mod tests {
     fn projection() {
         let mut item = Builder::new("an-id")
             .asset("data", "assets/dataset_geo.tif")
-            .into_item()
+            .build()
             .unwrap();
         super::update_item(&mut item, false, true).unwrap();
         let projection: Projection = item.extension().unwrap().unwrap();
diff --git a/core/src/geoparquet.rs b/core/src/geoparquet.rs
index 2ad323fe..53900cb4 100644
--- a/core/src/geoparquet.rs
+++ b/core/src/geoparquet.rs
@@ -43,9 +43,9 @@ mod has_feature {
     ///
     /// let item: Item = stac::read("examples/simple-item.json").unwrap();
     /// let mut cursor = Cursor::new(Vec::new());
-    /// stac::geoparquet::to_writer(&mut cursor, item.into()).unwrap();
+    /// stac::geoparquet::to_writer(&mut cursor, item).unwrap();
     /// ```
-    pub fn to_writer<W>(writer: W, value: Value) -> Result<()>
+    pub fn to_writer<W>(writer: W, value: impl Into<Value>) -> Result<()>
     where
         W: Write + Send,
     {
@@ -68,16 +68,17 @@ mod has_feature {
     /// let mut options = GeoParquetWriterOptions::default();
     /// let writer_properties = WriterProperties::builder().set_compression(Compression::SNAPPY).build();
     /// options.writer_properties = Some(writer_properties);
-    /// stac::geoparquet::to_writer_with_options(&mut cursor, item.into(), &options).unwrap();
+    /// stac::geoparquet::to_writer_with_options(&mut cursor, item, &options).unwrap();
     /// ```
     pub fn to_writer_with_options<W>(
         writer: W,
-        value: Value,
+        value: impl Into<Value>,
         options: &GeoParquetWriterOptions,
     ) -> Result<()>
     where
         W: Write + Send,
     {
+        let value = value.into();
         match value {
             Value::ItemCollection(item_collection) => {
                 let table = crate::geoarrow::to_table(item_collection)?;
@@ -88,7 +89,7 @@ mod has_feature {
                 )
                 .map_err(Error::from)
             }
-            Value::Item(item) => to_writer(writer, ItemCollection::from(vec![item.clone()]).into()),
+            Value::Item(item) => to_writer(writer, ItemCollection::from(vec![item])),
             _ => Err(Error::IncorrectType {
                 actual: value.type_name().to_string(),
                 expected: "Item or ItemCollection".to_string(),
@@ -118,21 +119,21 @@ mod has_feature {
 
     #[cfg(test)]
     mod tests {
-        use crate::{Href, Item, ItemCollection};
+        use crate::{Href, Item, ItemCollection, Value};
         use bytes::Bytes;
         use std::{fs::File, io::Cursor};
 
         #[test]
         fn to_writer_catalog() {
             let mut cursor = Cursor::new(Vec::new());
-            let catalog = crate::read("examples/catalog.json").unwrap();
+            let catalog: Value = crate::read("examples/catalog.json").unwrap();
             let _ = super::to_writer(&mut cursor, catalog).unwrap_err();
         }
 
         #[test]
         fn to_writer_collection() {
             let mut cursor = Cursor::new(Vec::new());
-            let collection = crate::read("examples/collection.json").unwrap();
+            let collection: Value = crate::read("examples/collection.json").unwrap();
             let _ = super::to_writer(&mut cursor, collection).unwrap_err();
         }
 
@@ -141,13 +142,13 @@ mod has_feature {
             let mut cursor = Cursor::new(Vec::new());
             let item = crate::read("examples/simple-item.json").unwrap();
             let item_collection = ItemCollection::from(vec![item]);
-            super::to_writer(&mut cursor, item_collection.into()).unwrap();
+            super::to_writer(&mut cursor, item_collection).unwrap();
         }
 
         #[test]
         fn to_writer_item() {
             let mut cursor = Cursor::new(Vec::new());
-            let item = crate::read("examples/simple-item.json").unwrap();
+            let item: Value = crate::read("examples/simple-item.json").unwrap();
             super::to_writer(&mut cursor, item).unwrap();
         }
 
@@ -163,7 +164,7 @@ mod has_feature {
             let mut item: Item = crate::read("examples/simple-item.json").unwrap();
             item.clear_href();
             let mut cursor = Cursor::new(Vec::new());
-            super::to_writer(&mut cursor, item.clone().into()).unwrap();
+            super::to_writer(&mut cursor, item.clone()).unwrap();
             let bytes = Bytes::from(cursor.into_inner());
             let item_collection = super::from_reader(bytes).unwrap();
             assert_eq!(item_collection.items[0], item);
diff --git a/core/src/io/json.rs b/core/src/io/json.rs
index ce620774..69de9c9c 100644
--- a/core/src/io/json.rs
+++ b/core/src/io/json.rs
@@ -5,7 +5,7 @@ use crate::{Error, Href, Result};
 #[cfg(feature = "reqwest")]
 use reqwest::blocking::Response;
 use serde::de::DeserializeOwned;
-use std::fs::File;
+use std::{fs::File, io::BufReader};
 
 /// Reads any STAC value from a JSON href.
 ///
@@ -22,7 +22,7 @@ struct JsonReader;
 
 impl<T: Href + DeserializeOwned> Read<T> for JsonReader {
     fn read_from_file(file: File) -> Result<T> {
-        serde_json::from_reader(file).map_err(Error::from)
+        serde_json::from_reader(BufReader::new(file)).map_err(Error::from)
     }
     #[cfg(feature = "reqwest")]
     fn from_response(response: Response) -> Result<T> {
diff --git a/core/src/item.rs b/core/src/item.rs
index c947fe9d..209a6660 100644
--- a/core/src/item.rs
+++ b/core/src/item.rs
@@ -306,17 +306,17 @@ impl Builder {
         self
     }
 
-    /// Creates an [Item] by consuming this builder.
+    /// Builds an [Item] from this builder.
     ///
     /// # Examples
     ///
     /// ```
     /// use stac::item::Builder;
     /// let builder = Builder::new("an-id").asset("data", "assets/dataset.tif");
-    /// let item = builder.into_item().unwrap();
+    /// let item = builder.build().unwrap();
     /// assert_eq!(item.assets.len(), 1);
     /// ```
-    pub fn into_item(self) -> Result<Item> {
+    pub fn build(self) -> Result<Item> {
         let mut item = Item::new(self.id);
         for (key, mut asset) in self.assets {
             if Url::parse(&asset.href).is_err() && self.canonicalize_paths {
@@ -889,7 +889,7 @@ mod tests {
     #[test]
     fn builder() {
         let builder = Builder::new("an-id").asset("data", "assets/dataset.tif");
-        let item = builder.into_item().unwrap();
+        let item = builder.build().unwrap();
         assert_eq!(item.assets.len(), 1);
         let asset = item.assets.get("data").unwrap();
         assert!(asset
@@ -902,7 +902,7 @@ mod tests {
         let builder = Builder::new("an-id")
             .canonicalize_paths(false)
             .asset("data", "assets/dataset.tif");
-        let item = builder.into_item().unwrap();
+        let item = builder.build().unwrap();
         let asset = item.assets.get("data").unwrap();
         assert_eq!(asset.href, "assets/dataset.tif");
     }
@@ -911,7 +911,7 @@ mod tests {
     fn builder_asset_roles() {
         let item = Builder::new("an-id")
             .asset("data", Asset::new("assets/dataset.tif").role("data"))
-            .into_item()
+            .build()
             .unwrap();
         let asset = item.assets.get("data").unwrap();
         assert_eq!(asset.roles, vec!["data"]);
@@ -921,7 +921,7 @@ mod tests {
     fn builder_uses_gdal() {
         let item = Builder::new("an-id")
             .asset("data", "assets/dataset.tif")
-            .into_item()
+            .build()
             .unwrap();
         if cfg!(feature = "gdal") {
             assert!(item.has_extension::<Raster>());
diff --git a/core/src/item_collection.rs b/core/src/item_collection.rs
index 4d11e837..51074d0d 100644
--- a/core/src/item_collection.rs
+++ b/core/src/item_collection.rs
@@ -1,6 +1,7 @@
 use crate::{Href, Item, Link, Links, Migrate};
 use serde::{Deserialize, Serialize};
 use serde_json::{Map, Value};
+use std::vec::IntoIter;
 
 /// The type field for [ItemCollections](ItemCollection).
 pub const ITEM_COLLECTION_TYPE: &str = "FeatureCollection";
@@ -55,6 +56,14 @@ impl FromIterator<Item> for ItemCollection {
     }
 }
 
+impl IntoIterator for ItemCollection {
+    type IntoIter = IntoIter<Item>;
+    type Item = Item;
+    fn into_iter(self) -> Self::IntoIter {
+        self.items.into_iter()
+    }
+}
+
 impl Href for ItemCollection {
     fn href(&self) -> Option<&str> {
         self.href.as_deref()
diff --git a/duckdb/Cargo.toml b/duckdb/Cargo.toml
index 10f4b952..28cdf25e 100644
--- a/duckdb/Cargo.toml
+++ b/duckdb/Cargo.toml
@@ -3,9 +3,6 @@ name = "stac-duckdb"
 version = "0.1.0"
 edition = "2021"
 
-[features]
-bundled = ["duckdb/bundled"]
-
 [dependencies]
 arrow = "52"
 duckdb = "1"
diff --git a/duckdb/src/lib.rs b/duckdb/src/lib.rs
index b723f44c..55fe9c82 100644
--- a/duckdb/src/lib.rs
+++ b/duckdb/src/lib.rs
@@ -259,26 +259,24 @@ impl Sql {
             params.extend(collections.into_iter().map(Value::from));
         }
         if let Some(intersects) = search.intersects {
-            wheres.push(format!(
-                "ST_Intersects(ST_GeomFromWKB(geometry), ST_GeomFromGeoJSON(?))"
-            ));
+            wheres
+                .push("ST_Intersects(ST_GeomFromWKB(geometry), ST_GeomFromGeoJSON(?))".to_string());
             params.push(Value::from(intersects.to_string()));
         }
         if let Some(bbox) = search.items.bbox {
-            wheres.push(format!(
-                "ST_Intersects(ST_GeomFromWKB(geometry), ST_GeomFromGeoJSON(?))"
-            ));
+            wheres
+                .push("ST_Intersects(ST_GeomFromWKB(geometry), ST_GeomFromGeoJSON(?))".to_string());
             params.push(Value::from(bbox.to_geometry().to_string()));
         }
         if let Some(datetime) = search.items.datetime {
             // TODO support start and end datetimes
             let (start, end) = stac::datetime::parse(&datetime)?;
             if let Some(start) = start {
-                wheres.push(format!("datetime >= make_timestamp(?)"));
+                wheres.push("datetime >= make_timestamp(?)".to_string());
                 params.push(Value::from(start.timestamp_micros()));
             }
             if let Some(end) = end {
-                wheres.push(format!("datetime <= make_timestamp(?)"));
+                wheres.push("datetime <= make_timestamp(?)".to_string());
                 params.push(Value::from(end.timestamp_micros()));
             }
         }
diff --git a/server/src/backend/mod.rs b/server/src/backend/mod.rs
index e7a67c18..ec0c4817 100644
--- a/server/src/backend/mod.rs
+++ b/server/src/backend/mod.rs
@@ -24,119 +24,6 @@ pub trait Backend: Clone + Sync + Send + 'static {
     /// ```
     fn has_item_search(&self) -> bool;
 
-    /// Adds collections and items from hrefs.
-    ///
-    /// A default implementation is provided. If `auto_create_collections` is
-    /// true, then, _if_ there is no collection for one or more items, a
-    /// collection will be auto-created before adding the items. If
-    /// `follow_links` is true, then `item` links on collections will be
-    /// followed and added as well.
-    ///
-    /// # Examples
-    ///
-    /// ```
-    /// use stac_server::{MemoryBackend, Backend};
-    /// let mut backend = MemoryBackend::new();
-    /// # tokio_test::block_on(async {
-    /// backend.add_from_hrefs(vec![
-    ///     "tests/data/collection.json".to_string(),
-    ///     "tests/data/feature.geojson".to_string(),
-    /// ], false, false);
-    /// # })
-    /// ```
-    #[cfg(feature = "tokio")]
-    async fn add_from_hrefs(
-        &mut self,
-        hrefs: Vec<String>,
-        auto_create_collections: bool,
-        follow_links: bool,
-    ) -> Result<()> {
-        use crate::Error;
-        use stac::{Href, Links, Value};
-        use std::collections::{HashMap, HashSet};
-        use tokio::task::JoinSet;
-
-        let mut set = JoinSet::new();
-        for href in hrefs {
-            // TODO allow parquet
-            let _ = set.spawn(async move { stac_async::read::<Value>(href).await });
-        }
-
-        let mut items: HashMap<Option<String>, Vec<Item>> = HashMap::new();
-        let mut item_collection_ids = HashSet::new();
-        let mut add_item = |mut item: Item| {
-            item.remove_structural_links();
-            if let Some(collection) = item.collection.as_ref() {
-                let collection = collection.clone();
-                let _ = item_collection_ids.insert(collection.clone());
-                let _ = items.entry(Some(collection)).or_default().push(item);
-            } else {
-                let _ = items.entry(None).or_default().push(item);
-            }
-        };
-        let mut item_set = JoinSet::new();
-        let mut collection_ids = HashSet::new();
-        while let Some(result) = set.join_next().await {
-            let value = result??;
-            match value {
-                Value::Item(item) => add_item(item),
-                Value::Catalog(catalog) => {
-                    return Err(Error::Backend(format!(
-                        "cannot add catalog with id={} to the backend",
-                        catalog.id
-                    )))
-                }
-                Value::Collection(mut collection) => {
-                    if follow_links {
-                        // TODO we could maybe merge this with `remove_structural_links`
-                        let href = collection
-                            .href()
-                            .expect("we read it, so it should have an href")
-                            .to_string();
-                        collection.make_relative_links_absolute(href)?;
-                        for link in collection.iter_item_links() {
-                            let href = link.href.clone();
-                            let _ =
-                                item_set.spawn(async move { stac_async::read::<Item>(href).await });
-                        }
-                    }
-                    collection.remove_structural_links();
-                    let _ = collection_ids.insert(collection.id.clone());
-                    self.add_collection(collection).await?
-                }
-                Value::ItemCollection(item_collection) => {
-                    for item in item_collection.items {
-                        add_item(item)
-                    }
-                }
-            }
-        }
-
-        while let Some(result) = item_set.join_next().await {
-            let item = result??;
-            add_item(item);
-        }
-
-        if auto_create_collections {
-            for id in item_collection_ids {
-                if !collection_ids.contains(&id) {
-                    let items = items
-                        .get(&Some(id.clone())) // TODO can we get rid of this clone?
-                        .expect("should have items for collection id");
-                    let collection = Collection::from_id_and_items(id, items);
-                    self.add_collection(collection).await?;
-                }
-            }
-        }
-
-        for (_, items) in items {
-            for item in items {
-                self.add_item(item).await?;
-            }
-        }
-        Ok(())
-    }
-
     /// Returns all collections.
     ///
     /// # Examples
@@ -251,28 +138,3 @@ pub trait Backend: Clone + Sync + Send + 'static {
     /// ```
     async fn search(&self, search: Search) -> Result<ItemCollection>;
 }
-
-#[cfg(test)]
-mod tests {
-    #[tokio::test]
-    #[cfg(feature = "tokio")]
-    async fn auto_create_collection() {
-        use super::Backend;
-        use crate::MemoryBackend;
-
-        let mut backend = MemoryBackend::new();
-        backend
-            .add_from_hrefs(
-                vec!["../spec-examples/v1.0.0/simple-item.json".to_string()],
-                true,
-                false,
-            )
-            .await
-            .unwrap();
-        let _ = backend
-            .collection("simple-collection")
-            .await
-            .unwrap()
-            .unwrap();
-    }
-}
diff --git a/validate/src/validator.rs b/validate/src/validator.rs
index 334c14ea..01d67353 100644
--- a/validate/src/validator.rs
+++ b/validate/src/validator.rs
@@ -4,6 +4,7 @@ use serde_json::Value;
 use stac::Version;
 use std::{
     collections::HashMap,
+    io::BufReader,
     sync::Arc,
     thread::{self, JoinHandle},
 };
@@ -305,7 +306,7 @@ impl SchemaResolver for Resolver {
                 "file" => {
                     if let Ok(path) = url.to_file_path() {
                         let f = std::fs::File::open(path)?;
-                        let document: Value = serde_json::from_reader(f)?;
+                        let document: Value = serde_json::from_reader(BufReader::new(f))?;
                         Ok(Arc::new(document))
                     } else {
                         Err(Error::InvalidFilePath(url.clone()).into())


Test patch:
diff --git a/cli/tests/item.rs b/cli/tests/item.rs
index 346c4b4a..e98f91cb 100644
--- a/cli/tests/item.rs
+++ b/cli/tests/item.rs
@@ -1,22 +1,7 @@
 use assert_cmd::Command;
-use stac::{Item, ItemCollection};
 
 #[test]
 fn item() {
     let mut command = Command::cargo_bin("stacrs").unwrap();
     command.arg("item").arg("an-id").assert().success();
 }
-
-#[test]
-fn item_collection() {
-    let mut command = Command::cargo_bin("stacrs").unwrap();
-    let item_a = serde_json::to_string(&Item::new("item-a")).unwrap();
-    let output = command
-        .arg("item")
-        .arg("item-b")
-        .arg("-c")
-        .write_stdin(item_a)
-        .unwrap();
-    let item_collection: ItemCollection = serde_json::from_slice(&output.stdout).unwrap();
-    assert_eq!(item_collection.items.len(), 2);
-}
diff --git a/cli/tests/sort.rs b/cli/tests/sort.rs
deleted file mode 100644
index b492053e..00000000
--- a/cli/tests/sort.rs
+++ /dev/null
@@ -1,13 +0,0 @@
-use assert_cmd::Command;
-use std::{fs::File, io::Read};
-
-#[test]
-fn sort_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("sort").write_stdin(item).assert().success();
-}