diff --git a/backend/Cargo.lock b/backend/Cargo.lock index 125def9e5b..2a0578646a 100644 --- a/backend/Cargo.lock +++ b/backend/Cargo.lock @@ -3147,6 +3147,7 @@ dependencies = [ "postgres-protocol", "serde", "serde_json", + "uuid 1.4.0", ] [[package]] diff --git a/backend/Cargo.toml b/backend/Cargo.toml index 680cbab638..08866dc684 100644 --- a/backend/Cargo.toml +++ b/backend/Cargo.toml @@ -178,6 +178,6 @@ serde-wasm-bindgen = "0.4" wasm-bindgen-test = "0.3.0" convert_case = "0.6.0" getrandom = "0.2" -tokio-postgres = {version = "^0.7", features = ["array-impls", "with-serde_json-1", "with-chrono-0_4"]} +tokio-postgres = {version = "^0.7", features = ["array-impls", "with-serde_json-1", "with-chrono-0_4", "with-uuid-1"]} postgres-native-tls = "^0" native-tls = "^0" \ No newline at end of file diff --git a/backend/windmill-worker/src/common.rs b/backend/windmill-worker/src/common.rs index a14afd1db8..ae7ea903c0 100644 --- a/backend/windmill-worker/src/common.rs +++ b/backend/windmill-worker/src/common.rs @@ -1,12 +1,5 @@ -use anyhow::{Context, Error}; -use chrono::{NaiveDateTime, Utc}; -use serde_json::Map; use sqlx::{Pool, Postgres}; use tokio::{fs::File, io::AsyncReadExt}; -use tokio_postgres::{ - types::{FromSql, Type}, - Column, Row, -}; use windmill_common::error::{self}; use windmill_queue::CLOUD_HOSTED; @@ -46,156 +39,3 @@ pub fn capitalize(s: &str) -> String { Some(f) => f.to_uppercase().collect::() + c.as_str(), } } - -pub fn postgres_row_to_json_value(row: Row) -> Result { - let row_data = postgres_row_to_row_data(row)?; - Ok(JSONValue::Object(row_data)) -} - -// some type-aliases I use in my project -pub type JSONValue = serde_json::Value; -pub type RowData = Map; - -pub fn postgres_row_to_row_data(row: Row) -> Result { - let mut result: Map = Map::new(); - for (i, column) in row.columns().iter().enumerate() { - let name = column.name(); - let json_value = pg_cell_to_json_value(&row, column, i)?; - result.insert(name.to_string(), json_value); - } - Ok(result) -} - -pub fn pg_cell_to_json_value( - row: &Row, - column: &Column, - column_i: usize, -) -> Result { - let f64_to_json_number = |raw_val: f64| -> Result { - let temp = serde_json::Number::from_f64(raw_val.into()) - .ok_or(anyhow::anyhow!("invalid json-float"))?; - Ok(JSONValue::Number(temp)) - }; - Ok(match *column.type_() { - // for rust-postgres <> postgres type-mappings: https://docs.rs/postgres/latest/postgres/types/trait.FromSql.html#types - // for postgres types: https://www.postgresql.org/docs/7.4/datatype.html#DATATYPE-TABLE - - // single types - Type::BOOL => get_basic(row, column, column_i, |a: bool| Ok(JSONValue::Bool(a)))?, - Type::INT2 => get_basic(row, column, column_i, |a: i16| { - Ok(JSONValue::Number(serde_json::Number::from(a))) - })?, - Type::INT4 => get_basic(row, column, column_i, |a: i32| { - Ok(JSONValue::Number(serde_json::Number::from(a))) - })?, - Type::INT8 => get_basic(row, column, column_i, |a: i64| { - Ok(JSONValue::Number(serde_json::Number::from(a))) - })?, - Type::TEXT | Type::VARCHAR => { - get_basic(row, column, column_i, |a: String| Ok(JSONValue::String(a)))? - } - Type::TIMESTAMP => get_basic(row, column, column_i, |a: chrono::NaiveDateTime| { - Ok(JSONValue::String(a.to_string())) - })?, - Type::TIMESTAMPTZ => get_basic(row, column, column_i, |a: chrono::DateTime| { - Ok(JSONValue::String(a.to_string())) - })?, - // Type::DATE => get_basic(row, column, column_i, |a: chrono::NaiveDate| { - // Ok(JSONValue::String(a.to_string())) - // })?, - Type::JSON | Type::JSONB => get_basic(row, column, column_i, |a: JSONValue| Ok(a))?, - Type::FLOAT4 => get_basic(row, column, column_i, |a: f32| { - Ok(f64_to_json_number(a.into())?) - })?, - Type::FLOAT8 => get_basic(row, column, column_i, |a: f64| Ok(f64_to_json_number(a)?))?, - // these types require a custom StringCollector struct as an intermediary (see struct at bottom) - Type::TS_VECTOR => get_basic(row, column, column_i, |a: StringCollector| { - Ok(JSONValue::String(a.0)) - })?, - - // array types - Type::BOOL_ARRAY => get_array(row, column, column_i, |a: bool| Ok(JSONValue::Bool(a)))?, - Type::INT2_ARRAY => get_array(row, column, column_i, |a: i16| { - Ok(JSONValue::Number(serde_json::Number::from(a))) - })?, - Type::INT4_ARRAY => get_array(row, column, column_i, |a: i32| { - Ok(JSONValue::Number(serde_json::Number::from(a))) - })?, - Type::INT8_ARRAY => get_array(row, column, column_i, |a: i64| { - Ok(JSONValue::Number(serde_json::Number::from(a))) - })?, - Type::TEXT_ARRAY | Type::VARCHAR_ARRAY => { - get_array(row, column, column_i, |a: String| Ok(JSONValue::String(a)))? - } - Type::JSON_ARRAY | Type::JSONB_ARRAY => { - get_array(row, column, column_i, |a: JSONValue| Ok(a))? - } - Type::FLOAT4_ARRAY => get_array(row, column, column_i, |a: f32| { - Ok(f64_to_json_number(a.into())?) - })?, - Type::FLOAT8_ARRAY => { - get_array(row, column, column_i, |a: f64| Ok(f64_to_json_number(a)?))? - } - // these types require a custom StringCollector struct as an intermediary (see struct at bottom) - Type::TS_VECTOR_ARRAY => get_array(row, column, column_i, |a: StringCollector| { - Ok(JSONValue::String(a.0)) - })?, - - _ => get_basic(row, column, column_i, |a: String| Ok(JSONValue::String(a)))?, - }) -} - -fn get_basic<'a, T: FromSql<'a>>( - row: &'a Row, - column: &Column, - column_i: usize, - val_to_json_val: impl Fn(T) -> Result, -) -> Result { - let raw_val = row.try_get::<_, Option>(column_i).with_context(|| { - format!( - "conversion issue for value at column_name:{}", - column.name() - ) - })?; - raw_val.map_or(Ok(JSONValue::Null), val_to_json_val) -} -fn get_array<'a, T: FromSql<'a>>( - row: &'a Row, - column: &Column, - column_i: usize, - val_to_json_val: impl Fn(T) -> Result, -) -> Result { - let raw_val_array = row - .try_get::<_, Option>>(column_i) - .with_context(|| { - format!( - "conversion issue for array at column_name:{}", - column.name() - ) - })?; - Ok(match raw_val_array { - Some(val_array) => { - let mut result = vec![]; - for val in val_array { - result.push(val_to_json_val(val)?); - } - JSONValue::Array(result) - } - None => JSONValue::Null, - }) -} - -// you can remove this section if not using TS_VECTOR (or other types requiring an intermediary `FromSQL` struct) -struct StringCollector(String); -impl FromSql<'_> for StringCollector { - fn from_sql( - _: &Type, - raw: &[u8], - ) -> Result> { - let result = std::str::from_utf8(raw)?; - Ok(StringCollector(result.to_owned())) - } - fn accepts(_ty: &Type) -> bool { - true - } -} diff --git a/backend/windmill-worker/src/pg_executor.rs b/backend/windmill-worker/src/pg_executor.rs index 7964318993..b4e407bf8a 100644 --- a/backend/windmill-worker/src/pg_executor.rs +++ b/backend/windmill-worker/src/pg_executor.rs @@ -1,19 +1,22 @@ +use anyhow::Context; +use chrono::Utc; use futures::TryStreamExt; use native_tls::TlsConnector; use postgres_native_tls::MakeTlsConnector; use serde::Deserialize; +use serde_json::Map; use serde_json::{json, Value}; use tokio_postgres::{types::ToSql, NoTls, Row}; -use windmill_common::{ - error::{to_anyhow, Error}, - jobs::QueuedJob, +use tokio_postgres::{ + types::{FromSql, Type}, + Column, }; + +use windmill_common::error::Error; +use windmill_common::{error::to_anyhow, jobs::QueuedJob}; use windmill_parser_sql::parse_sql_sig; -use crate::{ - common::postgres_row_to_json_value, get_content, transform_json_value, AuthedClient, - JobCompleted, -}; +use crate::{get_content, transform_json_value, AuthedClient, JobCompleted}; #[derive(Deserialize)] struct PgDatabase { @@ -171,3 +174,159 @@ pub async fn do_postgresql( // And then check that we got back the same string we sent over. return Ok(JobCompleted { job: job, result, logs: "".to_string(), success: true }); } + +pub fn pg_cell_to_json_value( + row: &Row, + column: &Column, + column_i: usize, +) -> Result { + let f64_to_json_number = |raw_val: f64| -> Result { + let temp = serde_json::Number::from_f64(raw_val.into()) + .ok_or(anyhow::anyhow!("invalid json-float"))?; + Ok(JSONValue::Number(temp)) + }; + Ok(match *column.type_() { + // for rust-postgres <> postgres type-mappings: https://docs.rs/postgres/latest/postgres/types/trait.FromSql.html#types + // for postgres types: https://www.postgresql.org/docs/7.4/datatype.html#DATATYPE-TABLE + + // single types + Type::BOOL => get_basic(row, column, column_i, |a: bool| Ok(JSONValue::Bool(a)))?, + Type::INT2 => get_basic(row, column, column_i, |a: i16| { + Ok(JSONValue::Number(serde_json::Number::from(a))) + })?, + Type::INT4 => get_basic(row, column, column_i, |a: i32| { + Ok(JSONValue::Number(serde_json::Number::from(a))) + })?, + Type::INT8 => get_basic(row, column, column_i, |a: i64| { + Ok(JSONValue::Number(serde_json::Number::from(a))) + })?, + Type::TEXT | Type::VARCHAR => { + get_basic(row, column, column_i, |a: String| Ok(JSONValue::String(a)))? + } + Type::TIMESTAMP => get_basic(row, column, column_i, |a: chrono::NaiveDateTime| { + Ok(JSONValue::String(a.to_string())) + })?, + Type::TIMESTAMPTZ => get_basic(row, column, column_i, |a: chrono::DateTime| { + Ok(JSONValue::String(a.to_string())) + })?, + Type::UUID => get_basic(row, column, column_i, |a: uuid::Uuid| { + Ok(JSONValue::String(a.to_string())) + })?, + // Type::DATE => get_basic(row, column, column_i, |a: chrono::NaiveDate| { + // Ok(JSONValue::String(a.to_string())) + // })?, + Type::JSON | Type::JSONB => get_basic(row, column, column_i, |a: JSONValue| Ok(a))?, + Type::FLOAT4 => get_basic(row, column, column_i, |a: f32| { + Ok(f64_to_json_number(a.into())?) + })?, + Type::FLOAT8 => get_basic(row, column, column_i, |a: f64| Ok(f64_to_json_number(a)?))?, + // these types require a custom StringCollector struct as an intermediary (see struct at bottom) + Type::TS_VECTOR => get_basic(row, column, column_i, |a: StringCollector| { + Ok(JSONValue::String(a.0)) + })?, + + // array types + Type::BOOL_ARRAY => get_array(row, column, column_i, |a: bool| Ok(JSONValue::Bool(a)))?, + Type::INT2_ARRAY => get_array(row, column, column_i, |a: i16| { + Ok(JSONValue::Number(serde_json::Number::from(a))) + })?, + Type::INT4_ARRAY => get_array(row, column, column_i, |a: i32| { + Ok(JSONValue::Number(serde_json::Number::from(a))) + })?, + Type::INT8_ARRAY => get_array(row, column, column_i, |a: i64| { + Ok(JSONValue::Number(serde_json::Number::from(a))) + })?, + Type::TEXT_ARRAY | Type::VARCHAR_ARRAY => { + get_array(row, column, column_i, |a: String| Ok(JSONValue::String(a)))? + } + Type::JSON_ARRAY | Type::JSONB_ARRAY => { + get_array(row, column, column_i, |a: JSONValue| Ok(a))? + } + Type::FLOAT4_ARRAY => get_array(row, column, column_i, |a: f32| { + Ok(f64_to_json_number(a.into())?) + })?, + Type::FLOAT8_ARRAY => { + get_array(row, column, column_i, |a: f64| Ok(f64_to_json_number(a)?))? + } + // these types require a custom StringCollector struct as an intermediary (see struct at bottom) + Type::TS_VECTOR_ARRAY => get_array(row, column, column_i, |a: StringCollector| { + Ok(JSONValue::String(a.0)) + })?, + _ => get_basic(row, column, column_i, |a: String| Ok(JSONValue::String(a)))?, + }) +} + +pub fn postgres_row_to_json_value(row: Row) -> Result { + let row_data = postgres_row_to_row_data(row)?; + Ok(JSONValue::Object(row_data)) +} + +// some type-aliases I use in my project +pub type JSONValue = serde_json::Value; +pub type RowData = Map; + +pub fn postgres_row_to_row_data(row: Row) -> Result { + let mut result: Map = Map::new(); + for (i, column) in row.columns().iter().enumerate() { + let name = column.name(); + let json_value = pg_cell_to_json_value(&row, column, i)?; + result.insert(name.to_string(), json_value); + } + Ok(result) +} + +fn get_basic<'a, T: FromSql<'a>>( + row: &'a Row, + column: &Column, + column_i: usize, + val_to_json_val: impl Fn(T) -> Result, +) -> Result { + let raw_val = row.try_get::<_, Option>(column_i).with_context(|| { + format!( + "conversion issue for value at column_name:{} with type {:?}", + column.name(), + column.type_() + ) + })?; + raw_val.map_or(Ok(JSONValue::Null), val_to_json_val) +} +fn get_array<'a, T: FromSql<'a>>( + row: &'a Row, + column: &Column, + column_i: usize, + val_to_json_val: impl Fn(T) -> Result, +) -> Result { + let raw_val_array = row + .try_get::<_, Option>>(column_i) + .with_context(|| { + format!( + "conversion issue for array at column_name:{}", + column.name() + ) + })?; + Ok(match raw_val_array { + Some(val_array) => { + let mut result = vec![]; + for val in val_array { + result.push(val_to_json_val(val)?); + } + JSONValue::Array(result) + } + None => JSONValue::Null, + }) +} + +// you can remove this section if not using TS_VECTOR (or other types requiring an intermediary `FromSQL` struct) +struct StringCollector(String); +impl FromSql<'_> for StringCollector { + fn from_sql( + _: &Type, + raw: &[u8], + ) -> Result> { + let result = std::str::from_utf8(raw)?; + Ok(StringCollector(result.to_owned())) + } + fn accepts(_ty: &Type) -> bool { + true + } +}