feat: crawl (#756)

## Closes

- Closes #750 as OBE

Author: gadomski

Cargo.toml

@@ -49,6 +49,8 @@ cql2 = "0.3.7"
 duckdb = "1.2.2"
 fluent-uri = "0.3.2"
 futures = "0.3.31"
+futures-core = "0.3.31"
+futures-util = "0.3.31"
 geo = "0.30.0"
 geo-traits = "0.2.0"
 geo-types = "0.7.15"
@@ -98,6 +100,7 @@ tracing-subscriber = { version = "0.3.18", features = [
     "env-filter",
     "tracing-log",
 ] }
+tracing-indicatif = "0.3.9"
 url = "2.3"
 webpki-roots = "1.0.0"
 

crates/cli/CHANGELOG.md

@@ -9,6 +9,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 ### Added
 
 - DuckDB server backend ([#651](https://github.com/stac-utils/rustac/pull/651))
+- Crawl ([#756](https://github.com/stac-utils/rustac/pull/756))
 
 ## [0.5.3] - 2025-02-20
 

crates/cli/Cargo.toml

@@ -18,8 +18,11 @@ duckdb-bundled = ["stac-duckdb/bundled"]
 
 [dependencies]
 anyhow.workspace = true
+async-stream.workspace = true
 axum.workspace = true
 clap = { workspace = true, features = ["derive"] }
+futures-core.workspace = true
+futures-util.workspace = true
 serde_json.workspace = true
 stac.workspace = true
 stac-api = { workspace = true, features = ["client"] }
@@ -38,7 +41,9 @@ tokio = { workspace = true, features = [
     "fs",
 ] }
 tracing.workspace = true
+tracing-indicatif.workspace = true
 tracing-subscriber = { workspace = true, features = ["env-filter"] }
+url.workspace = true
 
 [dev-dependencies]
 assert_cmd.workspace = true

crates/cli/src/lib.rs

@@ -3,15 +3,26 @@
 #![deny(unused_crate_dependencies)]
 
 use anyhow::{Error, Result, anyhow};
+use async_stream::try_stream;
 use clap::{Parser, Subcommand};
-use stac::{Collection, Href, Item, Links, Migrate, geoparquet::Compression};
+use futures_core::TryStream;
+use futures_util::{TryStreamExt, pin_mut};
+use stac::{Assets, Collection, Href, Item, Links, Migrate, SelfHref, geoparquet::Compression};
 use stac_api::{GetItems, GetSearch, Search};
-use stac_io::{Format, Validate};
+use stac_io::{Format, StacStore, Validate};
 use stac_server::Backend;
-use std::{collections::HashMap, io::Write, str::FromStr};
-use tokio::{io::AsyncReadExt, net::TcpListener, runtime::Handle};
+use std::{
+    collections::{HashMap, VecDeque},
+    io::Write,
+    str::FromStr,
+};
+use tokio::{io::AsyncReadExt, net::TcpListener, runtime::Handle, task::JoinSet};
 use tracing::metadata::Level;
-use tracing_subscriber::EnvFilter;
+use tracing_indicatif::IndicatifLayer;
+use tracing_subscriber::{
+    fmt::writer::MakeWriterExt, layer::SubscriberExt, util::SubscriberInitExt,
+};
+use url::Url;
 
 const DEFAULT_COLLECTION_ID: &str = "default-collection-id";
 
@@ -232,6 +243,19 @@ pub enum Command {
         create_collections: bool,
     },
 
+    /// Crawls a STAC Catalog or Collection by following its links.
+    ///
+    /// Items are saved as item collections (in the output format) in the output directory.
+    Crawl {
+        /// The href of a STAC Catalog or Collection
+        href: String,
+
+        /// The output directory
+        ///
+        /// This doesn't have to be local, by the way.
+        directory: String,
+    },
+
     /// Validates a STAC value.
     ///
     /// The default output format is plain text — use `--output-format=json` to
@@ -264,11 +288,16 @@ impl Rustac {
     /// is setting up the appropriate logging (e.g. Python).
     pub async fn run(self, init_tracing_subscriber: bool) -> Result<()> {
         if init_tracing_subscriber {
-            tracing_subscriber::fmt()
-                .with_env_filter(EnvFilter::from_default_env())
-                .with_max_level(self.log_level())
-                .with_writer(std::io::stderr)
-                .pretty()
+            let indicatif_layer = IndicatifLayer::new();
+            tracing_subscriber::registry()
+                .with(
+                    tracing_subscriber::fmt::layer().with_writer(
+                        indicatif_layer
+                            .get_stderr_writer()
+                            .with_max_level(self.log_level().unwrap_or(Level::WARN)),
+                    ),
+                )
+                .with(indicatif_layer)
                 .init();
         }
         match self.command {
@@ -418,6 +447,45 @@ impl Rustac {
                     load_and_serve(addr, backend, collections, items, create_collections).await
                 }
             }
+            Command::Crawl {
+                ref href,
+                ref directory,
+            } => {
+                let opts = self.opts();
+                let (store, path) = stac_io::parse_href_opts(href.clone(), opts.clone())?;
+                let value: stac::Value = store.get(path).await.unwrap();
+                let mut items: HashMap<Option<String>, Vec<Item>> = HashMap::new();
+                let crawl = crawl(value, store).await;
+                pin_mut!(crawl);
+                let mut warned = false;
+                while let Some(item) = crawl.try_next().await? {
+                    let collection = item.collection.clone();
+                    if collection.as_deref() == Some(DEFAULT_COLLECTION_ID) && !warned {
+                        warned = true;
+                        tracing::warn!(
+                            "collection id matches the default collection id, so any collection-less items will be grouped into this collection: {DEFAULT_COLLECTION_ID}"
+                        )
+                    }
+                    items.entry(collection).or_default().push(item);
+                }
+                let (store, path) = stac_io::parse_href_opts(directory.clone(), opts)?;
+                let format = self.output_format(None);
+                for (collection, items) in items {
+                    let file_name = format!(
+                        "{}.{}",
+                        collection.as_deref().unwrap_or(DEFAULT_COLLECTION_ID),
+                        format.extension()
+                    );
+                    store
+                        .put_format(
+                            path.child(file_name),
+                            stac::ItemCollection::from(items),
+                            format,
+                        )
+                        .await?;
+                }
+                Ok(())
+            }
             Command::Validate { ref infile } => {
                 let value = self.get(infile.as_deref()).await?;
                 let result = Handle::current()
@@ -651,6 +719,59 @@ fn level_value(level: Option<Level>) -> i8 {
     }
 }
 
+async fn crawl(value: stac::Value, store: StacStore) -> impl TryStream<Item = Result<Item>> {
+    use stac::Value::*;
+
+    try_stream! {
+        let mut values = VecDeque::from([value]);
+        while let Some(mut value) = values.pop_front() {
+            value.make_links_absolute()?;
+            match value {
+                Catalog(_) | Collection(_) => {
+                    if let Catalog(ref catalog) = value {
+                        tracing::info!("got catalog={}", catalog.id);
+                    }
+                    if let Collection(ref collection) = value {
+                        tracing::info!("got collection={}", collection.id);
+                    }
+                    let mut join_set: JoinSet<Result<stac::Value>> = JoinSet::new();
+                    for link in value
+                        .links()
+                        .iter()
+                        .filter(|link| link.is_child() || link.is_item())
+                        .cloned()
+                    {
+                        let store = store.clone();
+                        let url = Url::try_from(link.href)?;
+                        join_set.spawn(async move {
+                            let value: stac::Value = store.get(url.path()).await?;
+                            Ok(value)
+                        });
+                    }
+                    while let Some(result) = join_set.join_next().await {
+                        let value = result??;
+                        values.push_back(value);
+                    }
+                }
+                Item(mut item) => {
+                    if let Some(self_href) = item.self_href().cloned() {
+                        item.make_assets_absolute(self_href)?;
+                    }
+                    yield item;
+                }
+                ItemCollection(item_collection) => {
+                    for mut item in item_collection.items {
+                        if let Some(self_href) = item.self_href().cloned() {
+                            item.make_assets_absolute(self_href)?;
+                        }
+                        yield item;
+                    }
+                }
+            }
+        }
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use super::Rustac;

crates/core/CHANGELOG.md

@@ -12,10 +12,10 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 - `Clone` for `Container` ([#666](https://github.com/stac-utils/rustac/pull/666))
 - `Serialize` for `Container` ([#667](https://github.com/stac-utils/rustac/pull/667))
 - More permissive datetime interval parsing ([#715](https://github.com/stac-utils/rustac/pull/715))
-- `Format` methods for providing your own object store ([#730](https://github.com/stac-utils/rustac/pull/730))
 - `type` field to **stac-geoparquet** writes ([#736](https://github.com/stac-utils/rustac/pull/736))
 - `SelfHref::set_self_href` and `SelfHref::clear_self_href` ([#746](https://github.com/stac-utils/rustac/pull/746))
 - `Assets::make_assets_absolute` ([#753](https://github.com/stac-utils/rustac/pull/753))
+- `Format::extension` ([#756](https://github.com/stac-utils/rustac/pull/756))
 
 ### Changed
 
@@ -28,6 +28,10 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 
 - Support geometry columns other than "geometry" for **stac-geoparquet** ([#723](https://github.com/stac-utils/rustac/pull/723), [#727](https://github.com/stac-utils/rustac/pull/727))
 
+### Removed
+
+- IO (moved to **stac-io**) ([#739](https://github.com/stac-utils/rustac/pull/739))
+
 ## [0.12.0] - 2025-01-31
 
 ### Added

crates/io/src/format.rs

@@ -33,6 +33,26 @@ impl Format {
         href.rsplit_once('.').and_then(|(_, ext)| ext.parse().ok())
     }
 
+    /// Returns this format's file extension.
+    ///
+    /// # Examples
+    ///
+    /// ```
+    /// use stac_io::Format;
+    /// assert_eq!(Format::json().extension(), "json");
+    /// assert_eq!(Format::ndjson().extension(), "ndjson");
+    /// #[cfg(feature = "geoparquet")]
+    /// assert_eq!(Format::geoparquet().extension(), "parquet");
+    /// ```
+    pub fn extension(&self) -> &'static str {
+        match self {
+            Format::Json(_) => "json",
+            Format::NdJson => "ndjson",
+            #[cfg(feature = "geoparquet")]
+            Format::Geoparquet(_) => "parquet",
+        }
+    }
+
     /// Returns true if this is a geoparquet href.
     #[cfg(feature = "geoparquet")]
     pub fn is_geoparquet_href(href: &str) -> bool {

crates/io/src/store.rs

@@ -1,7 +1,8 @@
 use crate::{Format, Readable, Result, Writeable};
 use object_store::{ObjectStore, ObjectStoreScheme, PutResult, path::Path};
 use stac::Href;
-use std::sync::Arc;
+use std::{fmt::Debug, sync::Arc};
+use tracing::instrument;
 use url::Url;
 
 /// Parses an href into a [StacStore] and a [Path].
@@ -72,7 +73,7 @@ where
 }
 
 /// Reads STAC from an [ObjectStore].
-#[derive(Debug)]
+#[derive(Debug, Clone)]
 pub struct StacStore {
     store: Arc<dyn ObjectStore>,
     root: Url,
@@ -126,7 +127,8 @@ impl StacStore {
     }
 
     /// Gets a STAC value from the store in a specific format.
-    pub async fn get_format<T>(&self, path: impl Into<Path>, format: Format) -> Result<T>
+    #[instrument(skip(self))]
+    pub async fn get_format<T>(&self, path: impl Into<Path> + Debug, format: Format) -> Result<T>
     where
         T: Readable,
     {
@@ -141,22 +143,23 @@ impl StacStore {
     /// Puts a STAC value to the store.
     pub async fn put<T>(&self, path: impl Into<Path>, value: T) -> Result<PutResult>
     where
-        T: Writeable,
+        T: Writeable + Debug,
     {
         let path = path.into();
         let format = Format::infer_from_href(path.as_ref()).unwrap_or_default();
         self.put_format(path, value, format).await
     }
 
     /// Puts a STAC value to the store in a specific format.
+    #[instrument(skip(self))]
     pub async fn put_format<T>(
         &self,
-        path: impl Into<Path>,
+        path: impl Into<Path> + Debug,
         value: T,
         format: Format,
     ) -> Result<PutResult>
     where
-        T: Writeable,
+        T: Writeable + Debug,
     {
         let path = path.into();
         let bytes = format.into_vec(value)?;