diff --git a/Cargo.lock b/Cargo.lock index b6753e2d..d898d2ce 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1165,9 +1165,9 @@ checksum = "7f24254aa9a54b5c858eaee2f5bccdb46aaf0e486a595ed5fd8f86ba55232a70" [[package]] name = "hotdata" -version = "0.8.0" +version = "0.8.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "2e63f030edd9865312928059d492b3caf3c64f44b4a26abbfc92779517ec9026" +checksum = "6457e9ad27372a828225351937ef49a6a16ec789a2ec46b2a6fbfaa41b3bb780" dependencies = [ "arrow-array", "arrow-ipc", diff --git a/Cargo.toml b/Cargo.toml index f4e47c76..9ec0f00b 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -16,7 +16,7 @@ path = "src/main.rs" # behind a shared multi-thread tokio runtime. The `arrow` feature backs result # decode; `upload_file` runs the presigned direct-to-storage upload flow # (see src/sdk.rs::Api::upload). -hotdata = { version = "0.8.0", features = ["arrow"] } +hotdata = { version = "0.8.1", features = ["arrow"] } # Shared multi-thread runtime for the sync wrapper; block_on is called # concurrently from rayon worker threads (see src/indexes.rs). The presigned # upload (src/sdk.rs::Api::upload) drives the SDK's async `upload_file` on this diff --git a/README.md b/README.md index 2dcc3986..e6721c91 100644 --- a/README.md +++ b/README.md @@ -84,7 +84,7 @@ Pull data from SQL databases, APIs, S3/GCS/Azure buckets, or Iceberg catalogs into a managed database: ```sh -hotdata ingest connectors # browse: SQL dialects, ~150 API services, +hotdata ingest datasources # browse: SQL dialects, ~150 API services, # buckets, iceberg, api (bring-your-own) # Add a datasource — validates credentials and discovers the schema, loads no data. diff --git a/skills/hotdata/SKILL.md b/skills/hotdata/SKILL.md index 0566cdd6..07f4a2d3 100644 --- a/skills/hotdata/SKILL.md +++ b/skills/hotdata/SKILL.md @@ -302,8 +302,8 @@ Pull data from external sources (SQL databases, APIs, S3/GCS/Azure buckets, Iceb Commands that create or change things (`new-datasource`, `new-import`, `trigger-import`, `delete-datasource`) **require a workspace API key** (`HOTDATA_API_KEY` / `--api-key`, `hd_...`); a login session is not accepted. ```bash -hotdata ingest connectors [filter] # browse the catalog; "added" = you have a - # datasource for it. SQL dialects + ~150 named +hotdata ingest datasources [filter] # browse available datasource types; "added" = + # you already have one. SQL dialects + ~150 named # API services + generic buckets / iceberg / api. # -o json includes each entry's config_schema — # the exact fields --config takes. @@ -323,7 +323,7 @@ hotdata ingest new-datasource --service buckets --bucket-url s3://bucket/prefix hotdata ingest new-datasource --service iceberg --config @catalog.json --table ns.orders # Iceberg via a REST catalog. --table is REQUIRED (repeatable, namespace.table). # catalog.json fields: {"uri": …, "warehouse": …, "token": …} or "credential" + -# "scope" for an OAuth exchange — see the iceberg config_schema in `connectors -o json`. +# "scope" for an OAuth exchange — see the iceberg config_schema in `datasources -o json`. hotdata ingest list-datasources # names, ids, statuses; --all includes replaced ones hotdata ingest show-datasource # status + discovered tables/columns diff --git a/src/commands/databases.rs b/src/commands/databases.rs index d5eb2af8..98e2ad8b 100644 --- a/src/commands/databases.rs +++ b/src/commands/databases.rs @@ -258,6 +258,8 @@ struct DatabaseSummary { name: Option, #[serde(default)] default_catalog: Option, + #[serde(default)] + created_at: Option, } /// CLI output shape for `databases get`. A curated, stably-ordered view mapped @@ -275,6 +277,8 @@ pub struct Database { #[serde(default)] pub expires_at: Option, #[serde(default)] + pub created_at: Option, + #[serde(default)] attachments: Vec, } @@ -347,6 +351,7 @@ impl From for Database { default_catalog: Some(d.default_catalog), default_connection_id: d.default_connection_id, expires_at: d.expires_at.flatten(), + created_at: d.created_at.flatten(), attachments: d .attachments .into_iter() @@ -360,12 +365,12 @@ impl From for Database { } impl From for DatabaseSummary { - /// Map the SDK's typed summary into the CLI's list-row output shape. fn from(s: hotdata::models::DatabaseSummary) -> Self { DatabaseSummary { id: s.id, name: s.name.flatten(), default_catalog: Some(s.default_catalog), + created_at: s.created_at.flatten(), } } } @@ -386,7 +391,7 @@ pub(crate) fn get_database(api: &Api, id: &str) -> Result { /// up worker" hint instead of an unexplained pause — `databases list` is a /// common first command against an idle workspace. fn list_database_summaries(api: &Api) -> Result, ApiError> { - block_with_wakeup(api, "Loading databases…", api.client().databases().list()) + block_with_wakeup(api, "Loading databases...", api.client().databases().list()) .map(|r| r.databases.into_iter().map(DatabaseSummary::from).collect()) } @@ -398,8 +403,12 @@ fn list_database_summaries(api: &Api) -> Result, ApiError> /// rediscovers managed databases here and resolves each one's /// `default_connection_id` via [`get_database`]. The list summary omits the /// connection id, hence ids only. +/// +/// Deliberately spinner-less (plain [`block`], unlike +/// [`list_database_summaries`]): the caller owns its own "Loading indexes…" +/// spinner, and two indicatif bars would fight over the same line. pub(crate) fn list_database_ids(api: &Api) -> Result, ApiError> { - list_database_summaries(api).map(|dbs| dbs.into_iter().map(|d| d.id).collect()) + block(api.client().databases().list()).map(|r| r.databases.into_iter().map(|d| d.id).collect()) } fn fetch_database(api: &Api, id: &str) -> Database { @@ -818,10 +827,14 @@ pub fn list(workspace_id: &str, format: &str) { marker.to_string(), d.id.clone(), d.name.as_deref().unwrap_or("-").to_string(), + d.created_at + .as_deref() + .map(crate::util::format_date) + .unwrap_or_else(|| "-".to_string()), ] }) .collect(); - crate::output::table::print(&["", "ID", "NAME"], &rows); + crate::output::table::print(&["", "ID", "NAME", "CREATED"], &rows); } } _ => unreachable!(), @@ -845,6 +858,13 @@ pub fn get(workspace_id: &str, id_or_name: &str, format: &str) { if let Some(c) = &db.default_catalog { println!("{}{}", label("catalog:"), c.clone().cyan()); } + if let Some(ts) = &db.created_at { + println!( + "{}{}", + label("created_at:"), + crate::util::format_date(ts).dark_grey() + ); + } println!( "{}{}", label("default_connection_id:"), diff --git a/src/commands/indexes.rs b/src/commands/indexes.rs index 6bd24766..e7972bba 100644 --- a/src/commands/indexes.rs +++ b/src/commands/indexes.rs @@ -1,4 +1,4 @@ -use crate::client::sdk::{Api, block, block_with_wakeup, none_if_404}; +use crate::client::sdk::{Api, ApiError, block, block_with_wakeup, none_if_404}; use crate::commands::databases; use rayon::prelude::*; use serde::{Deserialize, Serialize}; @@ -182,8 +182,8 @@ fn connection_label_to_id_map(connections: &[ConnectionRef]) -> HashMap HashMap { - let resp = block(api.client().connections().list()).unwrap_or_else(|e| e.exit()); +fn connection_lookup(api: &Api) -> Result, ApiError> { + let resp = block(api.client().connections().list())?; let refs: Vec = resp .connections .into_iter() @@ -192,7 +192,7 @@ fn connection_lookup(api: &Api) -> HashMap { name: c.name, }) .collect(); - connection_label_to_id_map(&refs) + Ok(connection_label_to_id_map(&refs)) } /// Pick the connection id to address a per-table index call with during a @@ -233,15 +233,15 @@ struct ScanTarget { /// connection id, so each database needs a `databases get`; a database deleted /// between the list and the get (404) is skipped, any other error surfaces /// loudly to match the rest of this path. -fn managed_db_connection_ids(api: &Api) -> Vec { - let ids = databases::list_database_ids(api).unwrap_or_else(|e| e.exit()); - ids.par_iter() - .filter_map(|id| { - none_if_404(databases::get_database(api, id)) - .unwrap_or_else(|e| e.exit()) - .map(|db| db.default_connection_id) +fn managed_db_connection_ids(api: &Api) -> Result, ApiError> { + let ids = databases::list_database_ids(api)?; + let conn_ids: Result>, ApiError> = ids + .par_iter() + .map(|id| { + Ok(none_if_404(databases::get_database(api, id))?.map(|db| db.default_connection_id)) }) - .collect() + .collect(); + Ok(conn_ids?.into_iter().flatten().collect()) } /// Build the per-table scan list for a whole-workspace (unscoped) `indexes @@ -254,11 +254,15 @@ fn managed_db_connection_ids(api: &Api) -> Vec { /// with a connection-scoped `information_schema` call, exactly like the /// `--connection-id` path. The two table sets are disjoint: a managed database's /// connection is never returned by `connections list`. -fn workspace_scan_targets(api: &Api, schema: Option<&str>, table: Option<&str>) -> Vec { +fn workspace_scan_targets( + api: &Api, + schema: Option<&str>, + table: Option<&str>, +) -> Result, ApiError> { // Regular connections: one workspace-wide enumeration, label (= connection // name) mapped back to its id, falling back to the label itself (#161). - let name_to_id = connection_lookup(api); - let mut targets: Vec = collect_tables(api, None, schema, table) + let name_to_id = connection_lookup(api)?; + let mut targets: Vec = collect_tables(api, None, schema, table)? .into_iter() .map(|t| { let conn_id = scan_connection_id(None, &t.connection, &name_to_id).to_string(); @@ -268,21 +272,23 @@ fn workspace_scan_targets(api: &Api, schema: Option<&str>, table: Option<&str>) // Managed databases: discover their hidden connections, then scan each // scoped (the per-connection enumeration is what surfaces `__db_*` tables). - let db_conns = managed_db_connection_ids(api); - let managed: Vec = db_conns + let db_conns = managed_db_connection_ids(api)?; + let managed: Result>, ApiError> = db_conns .par_iter() - .flat_map(|conn| { - collect_tables(api, Some(conn), schema, table) - .into_iter() - .map(|t| ScanTarget { - conn_id: conn.clone(), - table: t, - }) - .collect::>() + .map(|conn| { + collect_tables(api, Some(conn), schema, table).map(|tables| { + tables + .into_iter() + .map(|t| ScanTarget { + conn_id: conn.clone(), + table: t, + }) + .collect::>() + }) }) .collect(); - targets.extend(managed); - targets + targets.extend(managed?.into_iter().flatten()); + Ok(targets) } /// Gather index rows across a connection's (or the workspace's) tables — the @@ -298,28 +304,28 @@ fn collect_connection_wide( connection_id: Option<&str>, schema: Option<&str>, table: Option<&str>, -) -> Vec { +) -> Result, ApiError> { let targets = match connection_id { - Some(cid) => collect_tables(api, Some(cid), schema, table) + Some(cid) => collect_tables(api, Some(cid), schema, table)? .into_iter() .map(|t| ScanTarget { conn_id: cid.to_string(), table: t, }) .collect(), - None => workspace_scan_targets(api, schema, table), + None => workspace_scan_targets(api, schema, table)?, }; - let per_table: Vec<(String, Vec)> = targets + let per_table: Result)>, ApiError> = targets .par_iter() .map(|tg| { let t = &tg.table; let full = format!("{}.{}.{}", t.connection, t.schema, t.table); - let indexes = list_one_table_scan(api, &tg.conn_id, &t.schema, &t.table); - (full, indexes) + let indexes = list_one_table_scan(api, &tg.conn_id, &t.schema, &t.table)?; + Ok((full, indexes)) }) .collect(); let mut rows: Vec = Vec::new(); - for (full, indexes) in per_table { + for (full, indexes) in per_table? { for i in indexes { rows.push(IndexRow { inner: i, @@ -327,7 +333,7 @@ fn collect_connection_wide( }); } } - rows + Ok(rows) } /// How to continue after merging one `/information_schema` page. @@ -358,7 +364,7 @@ fn collect_tables( connection_id: Option<&str>, schema: Option<&str>, table: Option<&str>, -) -> Vec { +) -> Result, ApiError> { let mut out = Vec::new(); let mut cursor: Option = None; loop { @@ -369,8 +375,7 @@ fn collect_tables( None, None, cursor.as_deref(), - )) - .unwrap_or_else(|e| e.exit()); + ))?; out.extend(resp.tables.into_iter().map(|t| InfoTable { connection: t.connection, schema: t.schema, @@ -383,23 +388,33 @@ fn collect_tables( } } sort_info_tables(&mut out); - out + Ok(out) } -fn list_one_table(api: &Api, connection_id: &str, schema: &str, table: &str) -> Vec { +fn list_one_table( + api: &Api, + connection_id: &str, + schema: &str, + table: &str, +) -> Result, ApiError> { // The SDK's typed `IndexInfoResponse.status` is a closed `ready`/`pending` // enum; the CLI accepts any status string for display. Keep the CLI's own // tolerant deserialization via the seam's untyped GET escape hatch. let path = format!("/connections/{connection_id}/tables/{schema}/{table}/indexes"); - let body: ListResponse = api.get_json(&path, &[]).unwrap_or_else(|e| e.exit()); - body.indexes + let body: ListResponse = api.get_json(&path, &[])?; + Ok(body.indexes) } -fn list_one_table_scan(api: &Api, connection_id: &str, schema: &str, table: &str) -> Vec { +fn list_one_table_scan( + api: &Api, + connection_id: &str, + schema: &str, + table: &str, +) -> Result, ApiError> { let path = format!("/connections/{connection_id}/tables/{schema}/{table}/indexes"); - match none_if_404(api.get_json::(&path, &[])).unwrap_or_else(|e| e.exit()) { - Some(body) => body.indexes, - None => Vec::new(), + match none_if_404(api.get_json::(&path, &[]))? { + Some(body) => Ok(body.indexes), + None => Ok(Vec::new()), } } @@ -479,7 +494,7 @@ pub fn infer_for_search( let connection_id = crate::commands::connections::resolve_connection_id(&api, connection_name); // Fetch indexes for this table - let indexes = list_one_table(&api, &connection_id, schema, table); + let indexes = list_one_table(&api, &connection_id, schema, table).unwrap_or_else(|e| e.exit()); let location = format!("{}.{}.{}", connection_name, schema, table); match resolve_search_params(&indexes, hint_type, hint_column, &location) { @@ -500,9 +515,13 @@ pub fn list( ) { let api = Api::new(Some(workspace_id)); - let (rows, multi_table) = match (connection_id, schema, table) { - (Some(cid), Some(sch), Some(tbl)) => { - let indexes = list_one_table(&api, cid, sch, tbl); + // One spinner over the whole fetch — the unscoped path is a + // whole-workspace scan (many requests) that otherwise sits silent. + // The database discovery inside is deliberately spinner-less + // (databases::list_database_ids) so nothing fights for the line. + let spinner = crate::util::spinner("Loading indexes…"); + let result = match (connection_id, schema, table) { + (Some(cid), Some(sch), Some(tbl)) => list_one_table(&api, cid, sch, tbl).map(|indexes| { let rows: Vec = indexes .into_iter() .map(|i| IndexRow { @@ -511,12 +530,11 @@ pub fn list( }) .collect(); (rows, false) - } - _ => ( - collect_connection_wide(&api, connection_id, schema, table), - true, - ), + }), + _ => collect_connection_wide(&api, connection_id, schema, table).map(|rows| (rows, true)), }; + spinner.finish_and_clear(); + let (rows, multi_table) = result.unwrap_or_else(|e| e.exit()); match format { "json" => println!("{}", serde_json::to_string_pretty(&rows).unwrap()), @@ -872,7 +890,7 @@ mod tests { .create(); let api = Api::test_new(&server.url(), "k", Some("ws")); - let rows = collect_connection_wide(&api, Some("conn-real"), None, None); + let rows = collect_connection_wide(&api, Some("conn-real"), None, None).unwrap(); info.assert(); idx.assert(); assert_eq!(rows.len(), 1); @@ -954,7 +972,7 @@ mod tests { .create(); let api = Api::test_new(&server.url(), "k", Some("ws")); - let rows = collect_connection_wide(&api, None, None, None); + let rows = collect_connection_wide(&api, None, None, None).unwrap(); conns.assert(); info_ws.assert(); dbs.assert(); @@ -1053,7 +1071,7 @@ mod tests { .create(); let api = Api::test_new(&server.url(), "k", Some("ws")); - let mut rows = collect_connection_wide(&api, None, None, None); + let mut rows = collect_connection_wide(&api, None, None, None).unwrap(); conns.assert(); info_ws.assert(); reg_idx.assert(); @@ -1087,7 +1105,7 @@ mod tests { .create(); let api = Api::test_new(&server.url(), "k", Some("ws1")); - let tables = collect_tables(&api, None, None, None); + let tables = collect_tables(&api, None, None, None).unwrap(); mock.assert(); assert_eq!(tables.len(), 2); assert_eq!(tables[0].table, "a"); @@ -1107,7 +1125,7 @@ mod tests { .create(); let api = Api::test_new(&server.url(), "k", Some("ws")); - let rows = list_one_table_scan(&api, "cid", "sch", "tbl"); + let rows = list_one_table_scan(&api, "cid", "sch", "tbl").unwrap(); mock.assert(); assert!(rows.is_empty()); } @@ -1134,7 +1152,7 @@ mod tests { .create(); let api = Api::test_new(&server.url(), "k", None); - let rows = list_one_table(&api, "cid", "sch", "tbl"); + let rows = list_one_table(&api, "cid", "sch", "tbl").unwrap(); mock.assert(); assert_eq!(rows.len(), 1); assert_eq!(rows[0].index_name, "ix1"); @@ -1164,7 +1182,7 @@ mod tests { .create(); let api = Api::test_new(&server.url(), "k", None); - let rows = list_one_table(&api, "cid", "sch", "tbl"); + let rows = list_one_table(&api, "cid", "sch", "tbl").unwrap(); mock.assert(); assert_eq!(rows.len(), 1); assert_eq!(rows[0].status, "building"); @@ -1181,7 +1199,7 @@ mod tests { .create(); let api = Api::test_new(&server.url(), "k", None); - let rows = list_one_table_scan(&api, "x", "s", "t"); + let rows = list_one_table_scan(&api, "x", "s", "t").unwrap(); mock.assert(); assert!(rows.is_empty()); } diff --git a/src/commands/ingest.rs b/src/commands/ingest.rs index d2381c47..f530934c 100644 --- a/src/commands/ingest.rs +++ b/src/commands/ingest.rs @@ -3,7 +3,8 @@ //! Two nouns, named explicitly in every command: **datasources** (added, //! credentialed sources — schema discovered, no data loaded) and **imports** //! (managed DBs materialized from a datasource). Commands: `new-datasource`, -//! `list-datasources`, `connectors` (the catalog), `new-import` (--all or +//! `list-datasources`, `datasources` (the catalog of available types), +//! `new-import` (--all or //! SQL), `list-imports`, `trigger-import` (re-run: refresh an import's DB //! from source), `status` (one-shot or `--wait` attach). The 0.13 //! `*-connection` verbs survive as hidden aliases (the noun collided with @@ -91,7 +92,7 @@ pub enum IngestCommands { /// /// Interactive by default; pass `--service` with config flags to add /// non-interactively. Pull rows separately with `hotdata ingest - /// new-import`; browse connectors with `hotdata ingest connectors`. + /// new-import`; browse available types with `hotdata ingest datasources`. #[command(alias = "new-connection")] NewDatasource { #[command(flatten)] @@ -130,9 +131,10 @@ pub enum IngestCommands { all: bool, }, - /// Browse the connector catalog - Connectors { - /// Filter to connectors whose name contains this text + /// Browse the catalog of available datasource types + #[command(alias = "connectors")] + Datasources { + /// Filter to entries whose name contains this text name: Option, }, @@ -231,7 +233,7 @@ pub fn dispatch(workspace_id: &str, output: &str, command: IngestCommands) { delete_datasource(workspace_id, output, &id, keep_database) } IngestCommands::ListDatasources { all } => list_datasources(workspace_id, output, all), - IngestCommands::Connectors { name } => catalog_list(workspace_id, name.as_deref(), output), + IngestCommands::Datasources { name } => catalog_list(workspace_id, name.as_deref(), output), IngestCommands::NewImport { sql, all, @@ -623,7 +625,7 @@ pub struct CreateArgs { name: Option, /// Connector config as JSON (inline, @file.json, or @-). Field reference: - /// `hotdata ingest connectors -o json` (each entry's config_schema) + /// `hotdata ingest datasources -o json` (each entry's config_schema) #[arg(long)] config: Option, @@ -676,7 +678,7 @@ fn add_from_flags(workspace_id: &str, output: &str, args: CreateArgs, wait: &Wai let Some(service) = args.service.as_deref() else { eprintln!( "error: --service is required to add a datasource non-interactively. \ - Browse connectors with 'hotdata ingest connectors', or run \ + Browse available types with 'hotdata ingest datasources', or run \ 'hotdata ingest new-datasource' in a terminal for the guided wizard." ); std::process::exit(1); @@ -688,7 +690,7 @@ fn add_from_flags(workspace_id: &str, output: &str, args: CreateArgs, wait: &Wai .find(|c| c.name == service) .cloned() .unwrap_or_else(|| { - eprintln!("error: unknown connector '{service}'. Run 'hotdata ingest connectors'."); + eprintln!("error: unknown connector '{service}'. Run 'hotdata ingest datasources'."); std::process::exit(1); }); @@ -1512,7 +1514,7 @@ fn delete_datasource(workspace_id: &str, output: &str, id: &str, keep_database: // --- catalog fetch -------------------------------------------------------- fn fetch_catalog(client: &IngestClient) -> Vec { - with_spinner("loading connectors…", || client.connectors()).connectors + with_spinner("loading datasource types…", || client.connectors()).connectors } /// Bare `api` connector: build a minimal API config interactively @@ -1564,7 +1566,7 @@ fn filesystem_credentials() -> serde_json::Value { /// Iceberg catalog connection config: prompt for catalog URI + warehouse + /// token + namespace. Field reference: the `iceberg` entry's config_schema -/// in `hotdata ingest connectors -o json`. +/// in `hotdata ingest datasources -o json`. fn iceberg_catalog_config() -> serde_json::Value { let mut m = serde_json::Map::new(); m.insert("uri".into(), ask_text("Catalog URI:").into()); diff --git a/src/commands/usage.rs b/src/commands/usage.rs index 931ca9d5..8f4146a9 100644 --- a/src/commands/usage.rs +++ b/src/commands/usage.rs @@ -41,7 +41,10 @@ pub fn usage(workspace_id: &str, since: Option<&str>, format: &str) { let query: Vec<(&str, String)> = since .map(|s| vec![("since", s.to_string())]) .unwrap_or_default(); - let u: Usage = api.get_json("/usage", &query).unwrap_or_else(|e| e.exit()); + let spinner = crate::util::spinner("Loading usage…"); + let result = api.get_json("/usage", &query); + spinner.finish_and_clear(); + let u: Usage = result.unwrap_or_else(|e| e.exit()); match format { "json" => println!("{}", serde_json::to_string_pretty(&u).unwrap()),