fix: use cql2-json for pgstac (#513)

Author: gadomski

Cargo.toml

@@ -45,6 +45,7 @@ bb8-postgres = "0.8.1"
 bytes = "1.7"
 chrono = "0.4.38"
 clap = "4.5"
+cql2 = "0.3.0"
 duckdb = "=1.0.0"
 fluent-uri = "0.3.1"
 futures = "0.3.31"
@@ -60,7 +61,7 @@ mime = "0.3.17"
 mockito = "1.5"
 object_store = "0.11.0"
 openssl = { version = "0.10.68", features = ["vendored"] }
-openssl-src = "=300.3.1"                                      # joinked from https://github.com/iopsystems/rpc-perf/commit/705b290d2105af6f33150da04b217422c6d68701#diff-2e9d962a08321605940b5a657135052fbcef87b5e360662bb527c96d9a615542R41 to cross-compile Python
+openssl-src = "=300.3.1" # joinked from https://github.com/iopsystems/rpc-perf/commit/705b290d2105af6f33150da04b217422c6d68701#diff-2e9d962a08321605940b5a657135052fbcef87b5e360662bb527c96d9a615542R41 to cross-compile Python
 parquet = { version = "52.2", default-features = false }
 pgstac = { version = "0.2.1", path = "crates/pgstac" }
 pyo3 = "0.22.3"
@@ -89,6 +90,9 @@ tokio-test = "0.4.4"
 tower = "0.5.1"
 tower-http = "0.6.1"
 tracing = "0.1.40"
-tracing-subscriber = "0.3.18"
+tracing-subscriber = { version = "0.3.18", features = [
+    "env-filter",
+    "tracing-log",
+] }
 url = "2.3"
 webpki-roots = "0.26.6"

crates/api/Cargo.toml

@@ -25,6 +25,7 @@ geo = ["dep:geo", "stac/geo"]
 [dependencies]
 async-stream = { workspace = true, optional = true }
 chrono.workspace = true
+cql2.workspace = true
 futures = { workspace = true, optional = true }
 http = { workspace = true, optional = true }
 reqwest = { workspace = true, features = ["json"], optional = true }

crates/api/src/client.rs

@@ -220,12 +220,7 @@ impl Client {
         if let Some(headers) = headers.into() {
             request = request.headers(headers);
         }
-        let response = request.send().await?;
-        if !response.status().is_success() {
-            let status_code = response.status();
-            let text = response.text().await.ok().unwrap_or_default();
-            return Err(Error::Request { status_code, text });
-        }
+        let response = request.send().await?.error_for_status()?;
         response.json().await.map_err(Error::from)
     }
 
@@ -361,8 +356,12 @@ fn stream_pages(
 
 fn not_found_to_none<T>(result: Result<T>) -> Result<Option<T>> {
     let mut result = result.map(Some);
-    if let Err(Error::Request { status_code, .. }) = result {
-        if status_code == StatusCode::NOT_FOUND {
+    if let Err(Error::Reqwest(ref err)) = result {
+        if err
+            .status()
+            .map(|s| s == StatusCode::NOT_FOUND)
+            .unwrap_or_default()
+        {
             result = Ok(None);
         }
     }

crates/api/src/error.rs

@@ -21,6 +21,10 @@ pub enum Error {
     #[error(transparent)]
     ChronoParse(#[from] chrono::ParseError),
 
+    /// [cql2::Error]
+    #[error(transparent)]
+    Cql2(#[from] cql2::Error),
+
     /// [geojson::Error]
     #[error(transparent)]
     GeoJson(#[from] Box<geojson::Error>),
@@ -75,17 +79,6 @@ pub enum Error {
     #[cfg(feature = "client")]
     Reqwest(#[from] reqwest::Error),
 
-    /// A search error.
-    #[error("HTTP status error ({status_code}): {text}")]
-    #[cfg(feature = "client")]
-    Request {
-        /// The status code
-        status_code: reqwest::StatusCode,
-
-        /// The text of the server response.
-        text: String,
-    },
-
     /// A search has both bbox and intersects.
     #[error("search has bbox and intersects")]
     SearchHasBboxAndIntersects(Box<Search>),

crates/api/src/filter.rs

@@ -1,7 +1,8 @@
-use std::{convert::Infallible, str::FromStr};
-
+use crate::Result;
+use cql2::Expr;
 use serde::{Deserialize, Serialize};
 use serde_json::{Map, Value};
+use std::{convert::Infallible, str::FromStr};
 
 /// The language of the filter expression.
 #[derive(Clone, Debug, Serialize, Deserialize, PartialEq)]
@@ -16,6 +17,32 @@ pub enum Filter {
     Cql2Json(Map<String, Value>),
 }
 
+impl Filter {
+    /// Converts this filter to cql2-json.
+    pub fn into_cql2_json(self) -> Result<Filter> {
+        match self {
+            Filter::Cql2Json(_) => Ok(self),
+            Filter::Cql2Text(text) => {
+                let expr = cql2::parse_text(&text)?;
+                Ok(Filter::Cql2Json(serde_json::from_value(
+                    serde_json::to_value(expr)?,
+                )?))
+            }
+        }
+    }
+
+    /// Converts this filter to cql2-json.
+    pub fn into_cql2_text(self) -> Result<Filter> {
+        match self {
+            Filter::Cql2Text(_) => Ok(self),
+            Filter::Cql2Json(json) => {
+                let expr: Expr = serde_json::from_value(Value::Object(json))?;
+                Ok(Filter::Cql2Text(expr.to_text()?))
+            }
+        }
+    }
+}
+
 impl Default for Filter {
     fn default() -> Self {
         Filter::Cql2Json(Default::default())
@@ -24,7 +51,7 @@ impl Default for Filter {
 
 impl FromStr for Filter {
     type Err = Infallible;
-    fn from_str(s: &str) -> Result<Self, Self::Err> {
+    fn from_str(s: &str) -> std::result::Result<Self, Self::Err> {
         Ok(Filter::Cql2Text(s.to_string()))
     }
 }

crates/api/src/items.rs

@@ -38,7 +38,7 @@ pub struct Items {
     pub filter_crs: Option<String>,
 
     /// CQL2 filter expression.
-    #[serde(skip_serializing_if = "Option::is_none")]
+    #[serde(skip_serializing_if = "Option::is_none", flatten)]
     pub filter: Option<Filter>,
 
     /// Additional filtering based on properties.
@@ -291,6 +291,14 @@ impl Items {
             collections: Some(vec![collection_id.to_string()]),
         }
     }
+
+    /// Converts the filter to cql2-json, if it is set.
+    pub fn into_cql2_json(mut self) -> Result<Items> {
+        if let Some(filter) = self.filter {
+            self.filter = Some(filter.into_cql2_json()?);
+        }
+        Ok(self)
+    }
 }
 
 impl TryFrom<Items> for GetItems {

crates/api/src/search.rs

@@ -211,6 +211,12 @@ impl Search {
             Ok(true)
         }
     }
+
+    /// Converts this search's filter to cql2-json, if set.
+    pub fn into_cql2_json(mut self) -> Result<Search> {
+        self.items = self.items.into_cql2_json()?;
+        Ok(self)
+    }
 }
 
 impl TryFrom<Search> for GetSearch {

crates/cli/src/args.rs

@@ -70,6 +70,10 @@ pub struct Args {
 #[derive(Debug, clap::Subcommand, Clone)]
 #[allow(clippy::large_enum_variant)]
 pub enum Subcommand {
+    /// Interact with a pgstac database
+    #[cfg(feature = "pgstac")]
+    Pgstac(crate::subcommand::pgstac::Args),
+
     /// Search for STAC items
     Search(search::Args),
 
@@ -99,6 +103,8 @@ impl Args {
     /// Runs whatever these arguments say that we should run.
     pub async fn run(self) -> Result<()> {
         match &self.subcommand {
+            #[cfg(feature = "pgstac")]
+            Subcommand::Pgstac(args) => self.pgstac(args).await,
             Subcommand::Search(args) => self.search(args).await,
             Subcommand::Serve(args) => self.serve(args).await,
             Subcommand::Translate(args) => self.translate(args).await,

crates/cli/src/subcommand/mod.rs

@@ -1,3 +1,5 @@
+#[cfg(feature = "pgstac")]
+pub mod pgstac;
 pub mod search;
 pub mod serve;
 pub mod translate;

crates/cli/src/subcommand/pgstac.rs

@@ -0,0 +1,57 @@
+use crate::Result;
+use stac_server::PgstacBackend;
+
+#[derive(Debug, Clone, clap::Args)]
+pub struct Args {
+    /// The pgstac subcommand
+    #[command(subcommand)]
+    subcommand: Subcommand,
+}
+
+#[derive(clap::Subcommand, Debug, Clone)]
+pub enum Subcommand {
+    /// Loads data into the pgstac database
+    Load(LoadArgs),
+}
+
+#[derive(clap::Args, Debug, Clone)]
+pub struct LoadArgs {
+    /// The connection string.
+    dsn: String,
+
+    /// Hrefs to load into the database.
+    ///
+    /// If not provided or `-`, data will be read from standard input.
+    hrefs: Vec<String>,
+
+    /// Load in all "item" links on collections.
+    #[arg(short, long)]
+    load_collection_items: bool,
+
+    /// Auto-generate collections for any collection-less items.
+    #[arg(short, long)]
+    create_collections: bool,
+}
+
+impl crate::Args {
+    pub async fn pgstac(&self, args: &Args) -> Result<()> {
+        match &args.subcommand {
+            Subcommand::Load(load_args) => {
+                let mut backend = PgstacBackend::new_from_stringlike(&load_args.dsn).await?;
+                let load = self
+                    .load(
+                        &mut backend,
+                        load_args.hrefs.iter().map(|h| h.as_str()),
+                        load_args.load_collection_items,
+                        load_args.create_collections,
+                    )
+                    .await?;
+                eprintln!(
+                    "Loaded {} collection(s) and {} item(s)",
+                    load.collections, load.items
+                );
+                Ok(())
+            }
+        }
+    }
+}