starred/windmill
1
0
Fork 0
mirror of https://github.com/windmill-labs/windmill.git synced 2026-10-27 16:02:43 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'main' into fr/flow-triggers-v0

Browse Source
This commit is contained in:
Faton Ramadani
2024-09-25 18:38:16 +02:00
committed by GitHub
parent 50654c6174 c2ae4fc799
commit 2f25c243c0
24 changed files with 422 additions and 286 deletions
Download Patch File Download Diff File Expand all files Collapse all files
CHANGELOG.md
+12
View File
@@ -1,5 +1,17 @@
# Changelog
## [1.401.0](https://github.com/windmill-labs/windmill/compare/v1.400.0...v1.401.0) (2024-09-25)
### Features
* add return_last_result annotation to sql ([#4443](https://github.com/windmill-labs/windmill/issues/4443)) ([3ce5587](https://github.com/windmill-labs/windmill/commit/3ce5587faae3912ceedae4644732fa9704eb6d76))
### Bug Fixes
* fix flow rendering ([fd58e7e](https://github.com/windmill-labs/windmill/commit/fd58e7eb48c4fb66d199c33d0f8aaf2535485a2f))
## [1.400.0](https://github.com/windmill-labs/windmill/compare/v1.399.0...v1.400.0) (2024-09-25)
backend/Cargo.lock
Generated
+22 -22
View File
@@ -10426,7 +10426,7 @@ checksum = "712e227841d057c1ee1cd2fb22fa7e5a5461ae8e48fa2ca79ec42cfc1931183f"
[[package]]
name = "windmill"
version = "1.400.0"
version = "1.401.0"
dependencies = [
"anyhow",
"axum",
@@ -10468,7 +10468,7 @@ dependencies = [
[[package]]
name = "windmill-api"
version = "1.400.0"
version = "1.401.0"
dependencies = [
"anyhow",
"argon2",
@@ -10553,7 +10553,7 @@ dependencies = [
[[package]]
name = "windmill-api-client"
version = "1.400.0"
version = "1.401.0"
dependencies = [
"base64 0.21.7",
"chrono",
@@ -10571,7 +10571,7 @@ dependencies = [
[[package]]
name = "windmill-audit"
version = "1.400.0"
version = "1.401.0"
dependencies = [
"chrono",
"serde",
@@ -10584,7 +10584,7 @@ dependencies = [
[[package]]
name = "windmill-common"
version = "1.400.0"
version = "1.401.0"
dependencies = [
"anyhow",
"async-stream",
@@ -10628,7 +10628,7 @@ dependencies = [
[[package]]
name = "windmill-git-sync"
version = "1.400.0"
version = "1.401.0"
dependencies = [
"regex",
"rsmq_async",
@@ -10643,7 +10643,7 @@ dependencies = [
[[package]]
name = "windmill-indexer"
version = "1.400.0"
version = "1.401.0"
dependencies = [
"anyhow",
"bytes",
@@ -10664,7 +10664,7 @@ dependencies = [
[[package]]
name = "windmill-parser"
version = "1.400.0"
version = "1.401.0"
dependencies = [
"convert_case 0.6.0",
"serde",
@@ -10673,7 +10673,7 @@ dependencies = [
[[package]]
name = "windmill-parser-bash"
version = "1.400.0"
version = "1.401.0"
dependencies = [
"anyhow",
"lazy_static",
@@ -10685,7 +10685,7 @@ dependencies = [
[[package]]
name = "windmill-parser-go"
version = "1.400.0"
version = "1.401.0"
dependencies = [
"anyhow",
"gosyn",
@@ -10697,7 +10697,7 @@ dependencies = [
[[package]]
name = "windmill-parser-graphql"
version = "1.400.0"
version = "1.401.0"
dependencies = [
"anyhow",
"lazy_static",
@@ -10709,7 +10709,7 @@ dependencies = [
[[package]]
name = "windmill-parser-php"
version = "1.400.0"
version = "1.401.0"
dependencies = [
"anyhow",
"itertools 0.13.0",
@@ -10720,7 +10720,7 @@ dependencies = [
[[package]]
name = "windmill-parser-py"
version = "1.400.0"
version = "1.401.0"
dependencies = [
"anyhow",
"itertools 0.13.0",
@@ -10731,7 +10731,7 @@ dependencies = [
[[package]]
name = "windmill-parser-py-imports"
version = "1.400.0"
version = "1.401.0"
dependencies = [
"anyhow",
"async-recursion",
@@ -10749,7 +10749,7 @@ dependencies = [
[[package]]
name = "windmill-parser-rust"
version = "1.400.0"
version = "1.401.0"
dependencies = [
"anyhow",
"convert_case 0.6.0",
@@ -10766,7 +10766,7 @@ dependencies = [
[[package]]
name = "windmill-parser-sql"
version = "1.400.0"
version = "1.401.0"
dependencies = [
"anyhow",
"lazy_static",
@@ -10778,7 +10778,7 @@ dependencies = [
[[package]]
name = "windmill-parser-ts"
version = "1.400.0"
version = "1.401.0"
dependencies = [
"anyhow",
"lazy_static",
@@ -10796,7 +10796,7 @@ dependencies = [
[[package]]
name = "windmill-parser-wasm"
version = "1.400.0"
version = "1.401.0"
dependencies = [
"anyhow",
"getrandom 0.2.15",
@@ -10817,7 +10817,7 @@ dependencies = [
[[package]]
name = "windmill-parser-yaml"
version = "1.400.0"
version = "1.401.0"
dependencies = [
"anyhow",
"serde_json",
@@ -10827,7 +10827,7 @@ dependencies = [
[[package]]
name = "windmill-queue"
version = "1.400.0"
version = "1.401.0"
dependencies = [
"anyhow",
"async-recursion",
@@ -10860,7 +10860,7 @@ dependencies = [
[[package]]
name = "windmill-sql-datatype-parser-wasm"
version = "1.400.0"
version = "1.401.0"
dependencies = [
"wasm-bindgen",
"wasm-bindgen-test",
@@ -10870,7 +10870,7 @@ dependencies = [
[[package]]
name = "windmill-worker"
version = "1.400.0"
version = "1.401.0"
dependencies = [
"anyhow",
"async-recursion",
backend/Cargo.toml
+2 -2
View File
@@ -1,6 +1,6 @@
[package]
name = "windmill"
version = "1.400.0"
version = "1.401.0"
authors.workspace = true
edition.workspace = true
@@ -27,7 +27,7 @@ members = [
]
[workspace.package]
version = "1.400.0"
version = "1.401.0"
authors = ["Ruben Fiszel <ruben@windmill.dev>"]
edition = "2021"
backend/windmill-api/openapi.yaml
+1 -1
View File
@@ -1,7 +1,7 @@
openapi: "3.0.3"
info:
version: 1.400.0
version: 1.401.0
title: Windmill API
contact:
backend/windmill-common/src/worker.rs
+16
View File
@@ -327,6 +327,22 @@ pub fn get_annotation(inner_content: &str) -> Annotations {
Annotations { npm_mode, nodejs_mode, native_mode, nobundling }
}
pub struct SqlAnnotations {
pub return_last_result: bool,
}
pub fn get_sql_annotations(inner_content: &str) -> SqlAnnotations {
let annotations = inner_content
.lines()
.take_while(|x| x.starts_with("--"))
.map(|x| x.to_string().replace("--", "").trim().to_string())
.collect_vec();
let return_last_result: bool = annotations.contains(&"return_last_result".to_string());
SqlAnnotations { return_last_result }
}
pub async fn load_cache(bin_path: &str, _remote_path: &str) -> (bool, String) {
if tokio::fs::metadata(&bin_path).await.is_ok() {
(true, format!("loaded from local cache: {}\n", bin_path))
backend/windmill-worker/src/bigquery_executor.rs
+85 -69
View File
@@ -5,6 +5,7 @@ use futures::{FutureExt, TryFutureExt};
use serde_json::{json, value::RawValue, Value};
use windmill_common::error::to_anyhow;
use windmill_common::jobs::QueuedJob;
use windmill_common::worker::get_sql_annotations;
use windmill_common::{error::Error, worker::to_raw_value};
use windmill_parser_sql::{
parse_bigquery_sig, parse_db_resource, parse_sql_blocks, parse_sql_statement_named_params,
@@ -69,6 +70,7 @@ fn do_bigquery_inner<'a>(
token: &'a str,
timeout_ms: i32,
column_order: Option<&'a mut Option<Vec<String>>>,
skip_collect: bool,
) -> windmill_common::error::Result<BoxFuture<'a, windmill_common::error::Result<Box<RawValue>>>> {
let param_names = parse_sql_statement_named_params(query, '@');
@@ -106,76 +108,80 @@ fn do_bigquery_inner<'a>(
match response.error_for_status_ref() {
Ok(_) => {
let result = response.json::<BigqueryResponse>().await.map_err(|e| {
Error::ExecutionErr(format!(
"BigQuery API response could not be parsed: {}",
e.to_string()
))
})?;
if skip_collect {
return Ok(to_raw_value(&Value::Array(vec![])));
} else {
let result = response.json::<BigqueryResponse>().await.map_err(|e| {
Error::ExecutionErr(format!(
"BigQuery API response could not be parsed: {}",
e.to_string()
))
})?;
if !result.jobComplete {
return Err(Error::ExecutionErr(
"BigQuery API did not answer query in time".to_string(),
if !result.jobComplete {
return Err(Error::ExecutionErr(
"BigQuery API did not answer query in time".to_string(),
));
}
if result.rows.is_none() || result.rows.as_ref().unwrap().len() == 0 {
return Ok(serde_json::from_str("[]").unwrap());
}
if result.schema.is_none() {
return Err(Error::ExecutionErr(
"Incomplete response from BigQuery API".to_string(),
));
}
if result
.totalRows
.unwrap_or(json!(""))
.as_str()
.unwrap_or("")
.parse::<i64>()
.unwrap_or(0)
> 10000
{
return Err(Error::ExecutionErr(
"More than 10000 rows were requested, use LIMIT 10000 to limit the number of rows".to_string(),
));
}
if let Some(column_order) = column_order {
*column_order = Some(
result
.schema
.as_ref()
.unwrap()
.fields
.iter()
.map(|x| x.name.clone())
.collect::<Vec<String>>(),
);
}
let rows = result
.rows
.unwrap()
.iter()
.map(|row| {
let mut row_map = serde_json::Map::new();
row.f
.iter()
.zip(result.schema.as_ref().unwrap().fields.iter())
.for_each(|(field, schema)| {
row_map.insert(
schema.name.clone(),
parse_val(&field.v, &schema.r#type, &schema),
);
});
Value::from(row_map)
})
.collect::<Vec<_>>();
Ok(to_raw_value(&rows))
}
if result.rows.is_none() || result.rows.as_ref().unwrap().len() == 0 {
return Ok(serde_json::from_str("[]").unwrap());
}
if result.schema.is_none() {
return Err(Error::ExecutionErr(
"Incomplete response from BigQuery API".to_string(),
));
}
if result
.totalRows
.unwrap_or(json!(""))
.as_str()
.unwrap_or("")
.parse::<i64>()
.unwrap_or(0)
> 10000
{
return Err(Error::ExecutionErr(
"More than 10000 rows were requested, use LIMIT 10000 to limit the number of rows".to_string(),
));
}
if let Some(column_order) = column_order {
*column_order = Some(
result
.schema
.as_ref()
.unwrap()
.fields
.iter()
.map(|x| x.name.clone())
.collect::<Vec<String>>(),
);
}
let rows = result
.rows
.unwrap()
.iter()
.map(|row| {
let mut row_map = serde_json::Map::new();
row.f
.iter()
.zip(result.schema.as_ref().unwrap().fields.iter())
.for_each(|(field, schema)| {
row_map.insert(
schema.name.clone(),
parse_val(&field.v, &schema.r#type, &schema),
);
});
Value::from(row_map)
})
.collect::<Vec<_>>();
return Ok(to_raw_value(&rows));
}
Err(e) => match response.json::<BigqueryErrorResponse>().await {
Ok(bq_err) => Err(Error::ExecutionErr(format!(
@@ -230,6 +236,8 @@ pub async fn do_bigquery(
return Err(Error::BadRequest("Missing database argument".to_string()));
};
let annotations = get_sql_annotations(query);
let service_account = CustomServiceAccount::from_json(&database)
.map_err(|e| Error::ExecutionErr(e.to_string()))?;
@@ -306,7 +314,8 @@ pub async fn do_bigquery(
let result_f = if queries.len() > 1 {
let futures = queries
.iter()
.map(|x| {
.enumerate()
.map(|(i, x)| {
do_bigquery_inner(
x,
&statement_values,
@@ -314,17 +323,23 @@ pub async fn do_bigquery(
token.as_str(),
timeout_ms,
None,
annotations.return_last_result && i < queries.len() - 1,
)
})
.collect::<windmill_common::error::Result<Vec<_>>>()?;
let f = async {
let mut res: Vec<Box<RawValue>> = vec![];
for fut in futures {
let r = fut.await?;
res.push(r);
}
Ok(to_raw_value(&res))
if annotations.return_last_result && res.len() > 0 {
Ok(res.pop().unwrap())
} else {
Ok(to_raw_value(&res))
}
};
f.boxed()
@@ -336,6 +351,7 @@ pub async fn do_bigquery(
token.as_str(),
timeout_ms,
Some(column_order),
false,
)?
};
backend/windmill-worker/src/mssql_executor.rs
+30 -22
View File
@@ -1,6 +1,5 @@
use base64::{engine::general_purpose, Engine as _};
use chrono::{DateTime, NaiveDate, NaiveDateTime, NaiveTime, Utc};
use futures::TryFutureExt;
use regex::Regex;
use serde::Deserialize;
use serde_json::value::RawValue;
@@ -10,7 +9,7 @@ use tokio::net::TcpStream;
use tokio_util::compat::TokioAsyncWriteCompatExt;
use uuid::Uuid;
use windmill_common::error::{self, Error};
use windmill_common::worker::to_raw_value;
use windmill_common::worker::{get_sql_annotations, to_raw_value};
use windmill_common::{error::to_anyhow, jobs::QueuedJob};
use windmill_parser_sql::{parse_db_resource, parse_mssql_sig};
use windmill_queue::{append_logs, CanceledBy};
@@ -67,6 +66,8 @@ pub async fn do_mssql(
return Err(Error::BadRequest("Missing database argument".to_string()));
};
let annotations = get_sql_annotations(query);
let mut config = Config::new();
config.host(database.host);
@@ -124,37 +125,44 @@ pub async fn do_mssql(
// polled to the end before querying again. Using streams allows
// fetching data in an asynchronous manner, if needed.
let stream = prepared_query.query(&mut client).await.map_err(to_anyhow)?;
stream
.into_results()
.await
.map_err(to_anyhow)?
.into_iter()
.map(|rows| {
let result = rows
.into_iter()
.map(|row| row_to_json(row))
.collect::<Result<Vec<Map<String, Value>>, Error>>();
result
})
.collect::<Result<Vec<Vec<Map<String, Value>>>, Error>>()
let results = stream.into_results().await.map_err(to_anyhow)?;
let len = results.len();
let mut json_results = vec![];
for (i, statement_result) in results.into_iter().enumerate() {
if annotations.return_last_result && i < len - 1 {
continue;
}
let mut json_rows = vec![];
for row in statement_result {
let row = row_to_json(row)?;
json_rows.push(row);
}
json_results.push(json_rows);
}
if annotations.return_last_result && json_results.len() > 0 {
Ok(to_raw_value(&json_results.pop().unwrap()))
} else {
Ok(to_raw_value(&json_results))
}
};
let rows = run_future_with_polling_update_job_poller(
let raw_result = run_future_with_polling_update_job_poller(
job.id,
job.timeout,
db,
mem_peak,
canceled_by,
result_f.map_err(to_anyhow),
result_f,
worker_name,
&job.workspace_id,
)
.await?;
let r = to_raw_value(&rows);
*mem_peak = (r.get().len() / 1000) as i32;
*mem_peak = (raw_result.get().len() / 1000) as i32;
return Ok(to_raw_value(&rows));
Ok(raw_result)
}
fn json_value_to_sql<'a>(
@@ -221,7 +229,7 @@ fn json_value_to_sql<'a>(
Ok(())
}
fn row_to_json(row: Row) -> Result<Map<String, Value>, Error> {
fn row_to_json(row: Row) -> Result<Value, Error> {
let cols = row
.columns()
.iter()
@@ -231,7 +239,7 @@ fn row_to_json(row: Row) -> Result<Map<String, Value>, Error> {
for (col, val) in cols.iter().zip(row.into_iter()) {
map.insert(col.name().to_string(), sql_to_json_value(val)?);
}
Ok(map)
Ok(Value::Object(map))
}
fn value_or_null<T>(
backend/windmill-worker/src/mysql_executor.rs
+63 -29
View File
@@ -13,6 +13,7 @@ use tokio::sync::Mutex;
use windmill_common::{
error::{to_anyhow, Error},
jobs::QueuedJob,
worker::{get_sql_annotations, to_raw_value},
};
use windmill_parser_sql::{
parse_db_resource, parse_mysql_sig, parse_sql_blocks, parse_sql_statement_named_params,
@@ -40,7 +41,8 @@ pub fn do_mysql_inner<'a>(
all_statement_values: &Params,
conn: Arc<Mutex<mysql_async::Conn>>,
column_order: Option<&'a mut Option<Vec<String>>>,
) -> windmill_common::error::Result<BoxFuture<'a, anyhow::Result<Vec<Value>>>> {
skip_collect: bool,
) -> windmill_common::error::Result<BoxFuture<'a, anyhow::Result<Box<RawValue>>>> {
let param_names = parse_sql_statement_named_params(query, ':')
.into_iter()
.map(|x| x.into_bytes())
@@ -58,31 +60,42 @@ pub fn do_mysql_inner<'a>(
};
let result_f = async move {
let rows: Vec<Row> = conn
.lock()
.await
.exec(query, statement_values)
.await
.map_err(to_anyhow)?;
if skip_collect {
conn.lock()
.await
.exec_drop(query, statement_values)
.await
.map_err(to_anyhow)?;
if let Some(column_order) = column_order {
*column_order = Some(
rows.first()
.map(|x| {
x.columns()
.iter()
.map(|x| x.name_str().to_string())
.collect::<Vec<String>>()
})
.unwrap_or_default(),
);
Ok(to_raw_value(&Value::Array(vec![])))
} else {
let rows: Vec<Row> = conn
.lock()
.await
.exec(query, statement_values)
.await
.map_err(to_anyhow)?;
if let Some(column_order) = column_order {
*column_order = Some(
rows.first()
.map(|x| {
x.columns()
.iter()
.map(|x| x.name_str().to_string())
.collect::<Vec<String>>()
})
.unwrap_or_default(),
);
}
Ok(to_raw_value(
&rows
.into_iter()
.map(|x| convert_row_to_value(x))
.collect::<Vec<serde_json::Value>>(),
))
}
Ok(rows
.into_iter()
.map(|x| convert_row_to_value(x))
.collect::<Vec<serde_json::Value>>())
as Result<Vec<serde_json::Value>, anyhow::Error>
};
Ok(result_f.boxed())
@@ -133,6 +146,8 @@ pub async fn do_mysql(
return Err(Error::BadRequest("Missing database argument".to_string()));
};
let annotations = get_sql_annotations(query);
let opts = OptsBuilder::default()
.db_name(Some(database.database))
.user(database.user)
@@ -235,21 +250,40 @@ pub async fn do_mysql(
let result_f = if queries.len() > 1 {
let futures = queries
.iter()
.map(|x| do_mysql_inner(x, &statement_values, conn_a.clone(), None))
.enumerate()
.map(|(i, x)| {
do_mysql_inner(
x,
&statement_values,
conn_a.clone(),
None,
annotations.return_last_result && i < queries.len() - 1,
)
})
.collect::<windmill_common::error::Result<Vec<_>>>()?;
let f = async {
let mut res: Vec<serde_json::Value> = vec![];
let mut res: Vec<Box<RawValue>> = vec![];
for fut in futures {
let r = fut.await?;
res.push(serde_json::to_value(r).map_err(to_anyhow)?);
res.push(r);
}
if annotations.return_last_result && res.len() > 0 {
Ok(res.pop().unwrap())
} else {
Ok(to_raw_value(&res))
}
Ok(res)
};
f.boxed()
} else {
do_mysql_inner(query, &statement_values, conn_a.clone(), Some(column_order))?
do_mysql_inner(
query,
&statement_values,
conn_a.clone(),
Some(column_order),
false,
)?
};
let result = run_future_with_polling_update_job_poller(
backend/windmill-worker/src/pg_executor.rs
+64 -44
View File
@@ -30,7 +30,7 @@ use tokio_postgres::{
};
use uuid::Uuid;
use windmill_common::error::{self, Error};
use windmill_common::worker::{to_raw_value, CLOUD_HOSTED};
use windmill_common::worker::{get_sql_annotations, to_raw_value, CLOUD_HOSTED};
use windmill_common::{error::to_anyhow, jobs::QueuedJob};
use windmill_parser::{Arg, Typ};
use windmill_parser_sql::{
@@ -68,7 +68,8 @@ fn do_postgresql_inner<'a>(
client: &'a Client,
column_order: Option<&'a mut Option<Vec<String>>>,
siz: &'a AtomicUsize,
) -> error::Result<BoxFuture<'a, anyhow::Result<Vec<Value>>>> {
skip_collect: bool,
) -> error::Result<BoxFuture<'a, anyhow::Result<Box<RawValue>>>> {
let mut query_params = vec![];
let arg_indices = parse_pg_statement_arg_indices(&query);
@@ -93,51 +94,60 @@ fn do_postgresql_inner<'a>(
let result_f = async move {
// Now we can execute a simple statement that just returns its parameter.
let rows = client
.query_raw(&query, query_params)
.await
.map_err(to_anyhow)?;
let rows = rows.try_collect::<Vec<Row>>().await.map_err(to_anyhow)?;
if let Some(column_order) = column_order {
*column_order = Some(
rows.first()
.map(|x| {
x.columns()
.iter()
.map(|x| x.name().to_string())
.collect::<Vec<String>>()
})
.unwrap_or_default(),
);
}
let mut res: Vec<serde_json::Value> = vec![];
for row in rows.into_iter() {
let r = postgres_row_to_json_value(row);
if let Ok(v) = r.as_ref() {
let size = sizeof_val(v);
siz.fetch_add(size, Ordering::Relaxed);
if skip_collect {
client
.execute_raw(&query, query_params)
.await
.map_err(to_anyhow)?;
} else {
let rows = client
.query_raw(&query, query_params)
.await
.map_err(to_anyhow)?;
let rows = rows.try_collect::<Vec<Row>>().await.map_err(to_anyhow)?;
if let Some(column_order) = column_order {
*column_order = Some(
rows.first()
.map(|x| {
x.columns()
.iter()
.map(|x| x.name().to_string())
.collect::<Vec<String>>()
})
.unwrap_or_default(),
);
}
if *CLOUD_HOSTED {
let siz = siz.load(Ordering::Relaxed);
if siz > MAX_RESULT_SIZE * 4 {
return Err(anyhow::anyhow!(
"Query result too large for cloud (size = {} > {})",
siz,
MAX_RESULT_SIZE & 4
));
for row in rows.into_iter() {
let r = postgres_row_to_json_value(row);
if let Ok(v) = r.as_ref() {
let size = sizeof_val(v);
siz.fetch_add(size, Ordering::Relaxed);
}
if *CLOUD_HOSTED {
let siz = siz.load(Ordering::Relaxed);
if siz > MAX_RESULT_SIZE * 4 {
return Err(anyhow::anyhow!(
"Query result too large for cloud (size = {} > {})",
siz,
MAX_RESULT_SIZE & 4
));
}
}
if let Ok(v) = r {
res.push(v);
} else {
return Err(to_anyhow(r.err().unwrap()));
}
}
if let Ok(v) = r {
res.push(v);
} else {
return Err(to_anyhow(r.err().unwrap()));
}
}
Ok(res)
Ok(to_raw_value(&res))
};
Ok(result_f.boxed())
@@ -178,6 +188,9 @@ pub async fn do_postgresql(
} else {
return Err(Error::BadRequest("Missing database argument".to_string()));
};
let annotations = get_sql_annotations(query);
let sslmode = match database.sslmode.as_deref() {
Some("allow") => "prefer".to_string(),
Some("verify-ca") | Some("verify-full") => "require".to_string(),
@@ -291,24 +304,30 @@ pub async fn do_postgresql(
let result_f = if queries.len() > 1 {
let futures = queries
.iter()
.map(|x| {
.enumerate()
.map(|(i, x)| {
do_postgresql_inner(
x.to_string(),
&param_idx_to_arg_and_value,
client,
None,
&size,
annotations.return_last_result && i < queries.len() - 1,
)
})
.collect::<error::Result<Vec<_>>>()?;
let f = async {
let mut res: Vec<serde_json::Value> = vec![];
let mut res: Vec<Box<RawValue>> = vec![];
for fut in futures {
let r = fut.await?;
res.push(serde_json::to_value(r).map_err(to_anyhow)?);
res.push(r);
}
if annotations.return_last_result && res.len() > 0 {
Ok(res.pop().unwrap())
} else {
Ok(to_raw_value(&res))
}
Ok(res)
};
f.boxed()
@@ -319,6 +338,7 @@ pub async fn do_postgresql(
client,
Some(column_order),
&size,
false,
)?
};
backend/windmill-worker/src/snowflake_executor.rs
+108 -81
View File
@@ -9,6 +9,7 @@ use serde_json::{json, value::RawValue, Value};
use sha2::{Digest, Sha256};
use std::collections::HashMap;
use windmill_common::error::to_anyhow;
use windmill_common::worker::get_sql_annotations;
use windmill_common::jobs::QueuedJob;
use windmill_common::{error::Error, worker::to_raw_value};
@@ -74,34 +75,41 @@ struct SnowflakeError {
}
trait SnowflakeResponseExt {
async fn get_snowflake_response<T: for<'a> Deserialize<'a>>(
async fn parse_snowflake_response<T: for<'a> Deserialize<'a>>(
self,
) -> windmill_common::error::Result<T>;
}
async fn handle_snowflake_result(
result: Result<Response, reqwest::Error>,
) -> windmill_common::error::Result<Response> {
match result {
Ok(response) => match response.error_for_status_ref() {
Ok(_) => Ok(response),
Err(e) => {
let resp = response.text().await.unwrap_or("".to_string());
match serde_json::from_str::<SnowflakeError>(&resp) {
Ok(sf_err) => return Err(Error::ExecutionErr(sf_err.message)),
Err(_) => return Err(Error::ExecutionErr(e.to_string())),
}
}
},
Err(e) => Err(Error::ExecutionErr(format!(
"Could not send request: {:?}",
e
))),
}
}
impl SnowflakeResponseExt for Result<Response, reqwest::Error> {
async fn get_snowflake_response<T: for<'a> Deserialize<'a>>(
async fn parse_snowflake_response<T: for<'a> Deserialize<'a>>(
self,
) -> windmill_common::error::Result<T> {
match self {
Ok(response) => match response.error_for_status_ref() {
Ok(_) => response
.json::<T>()
.await
.map_err(|e| Error::ExecutionErr(e.to_string())),
Err(e) => {
let resp = response.text().await.unwrap_or("".to_string());
match serde_json::from_str::<SnowflakeError>(&resp) {
Ok(sf_err) => return Err(Error::ExecutionErr(sf_err.message)),
Err(_) => return Err(Error::ExecutionErr(e.to_string())),
}
}
},
Err(e) => Err(Error::ExecutionErr(format!(
"Could not send request: {:?}",
e
))),
}
let response = handle_snowflake_result(self).await?;
response
.json::<T>()
.await
.map_err(|e| Error::ExecutionErr(e.to_string()))
}
}
@@ -112,6 +120,7 @@ fn do_snowflake_inner<'a>(
account_identifier: &'a str,
token: &'a str,
column_order: Option<&'a mut Option<Vec<String>>>,
skip_collect: bool,
) -> windmill_common::error::Result<BoxFuture<'a, windmill_common::error::Result<Box<RawValue>>>> {
body.insert("statement".to_string(), json!(query));
@@ -135,7 +144,7 @@ fn do_snowflake_inner<'a>(
}
let result_f = async move {
let response = HTTP_CLIENT
let result = HTTP_CLIENT
.post(format!(
"https://{}.snowflakecomputing.com/api/v2/statements/",
account_identifier.to_uppercase()
@@ -144,67 +153,76 @@ fn do_snowflake_inner<'a>(
.header("X-Snowflake-Authorization-Token-Type", "KEYPAIR_JWT")
.json(&body)
.send()
.await
.get_snowflake_response::<SnowflakeResponse>()
.await?;
.await;
if response.resultSetMetaData.numRows > 10000 {
return Err(Error::ExecutionErr(
"More than 10000 rows were requested, use LIMIT 10000 to limit the number of rows"
.to_string(),
));
}
if let Some(column_order) = column_order {
*column_order = Some(
response
.resultSetMetaData
.rowType
.iter()
.map(|x| x.name.clone())
.collect::<Vec<String>>(),
);
}
if skip_collect {
handle_snowflake_result(result).await?;
Ok(to_raw_value(&Value::Array(vec![])))
} else {
let response = result
.parse_snowflake_response::<SnowflakeResponse>()
.await?;
let mut rows = response.data;
if response.resultSetMetaData.partitionInfo.len() > 1 {
for idx in 1..response.resultSetMetaData.partitionInfo.len() {
let url = format!(
"https://{}.snowflakecomputing.com/api/v2/statements/{}",
account_identifier.to_uppercase(),
response.statementHandle
);
let response = HTTP_CLIENT
.get(url)
.bearer_auth(token)
.header("X-Snowflake-Authorization-Token-Type", "KEYPAIR_JWT")
.query(&[("partition", idx.to_string())])
.send()
.await
.get_snowflake_response::<SnowflakeDataOnlyResponse>()
.await?;
rows.extend(response.data);
if response.resultSetMetaData.numRows > 10000 {
return Err(Error::ExecutionErr(
"More than 10000 rows were requested, use LIMIT 10000 to limit the number of rows"
.to_string(),
));
}
if let Some(column_order) = column_order {
*column_order = Some(
response
.resultSetMetaData
.rowType
.iter()
.map(|x| x.name.clone())
.collect::<Vec<String>>(),
);
}
let mut rows = response.data;
if response.resultSetMetaData.partitionInfo.len() > 1 {
for idx in 1..response.resultSetMetaData.partitionInfo.len() {
let url = format!(
"https://{}.snowflakecomputing.com/api/v2/statements/{}",
account_identifier.to_uppercase(),
response.statementHandle
);
let response = HTTP_CLIENT
.get(url)
.bearer_auth(token)
.header("X-Snowflake-Authorization-Token-Type", "KEYPAIR_JWT")
.query(&[("partition", idx.to_string())])
.send()
.await
.parse_snowflake_response::<SnowflakeDataOnlyResponse>()
.await?;
rows.extend(response.data);
}
}
let rows = to_raw_value(
&rows
.iter()
.map(|row| {
let mut row_map = serde_json::Map::new();
row.iter()
.zip(response.resultSetMetaData.rowType.iter())
.for_each(|(val, row_type)| {
row_map.insert(
row_type.name.clone(),
parse_val(&val, &row_type.r#type),
);
});
row_map
})
.collect::<Vec<_>>(),
);
Ok(rows)
}
let rows = to_raw_value(
&rows
.iter()
.map(|row| {
let mut row_map = serde_json::Map::new();
row.iter()
.zip(response.resultSetMetaData.rowType.iter())
.for_each(|(val, row_type)| {
row_map
.insert(row_type.name.clone(), parse_val(&val, &row_type.r#type));
});
row_map
})
.collect::<Vec<_>>(),
);
Ok(rows)
};
Ok(result_f.boxed())
@@ -246,6 +264,8 @@ pub async fn do_snowflake(
return Err(Error::BadRequest("Missing database argument".to_string()));
};
let annotations = get_sql_annotations(query);
let qualified_username = format!(
"{}.{}",
database.account_identifier.split('.').next().unwrap_or(""), // get first part of account identifier
@@ -315,7 +335,8 @@ pub async fn do_snowflake(
let result_f = if queries.len() > 1 {
let futures = queries
.iter()
.map(|x| {
.enumerate()
.map(|(i, x)| {
do_snowflake_inner(
x,
&snowflake_args,
@@ -323,6 +344,7 @@ pub async fn do_snowflake(
&database.account_identifier,
&token,
None,
annotations.return_last_result && i < queries.len() - 1,
)
})
.collect::<windmill_common::error::Result<Vec<_>>>()?;
@@ -333,7 +355,11 @@ pub async fn do_snowflake(
let r = fut.await?;
res.push(r);
}
Ok(to_raw_value(&res))
if annotations.return_last_result && res.len() > 0 {
Ok(res.pop().unwrap())
} else {
Ok(to_raw_value(&res))
}
};
f.boxed()
@@ -345,6 +371,7 @@ pub async fn do_snowflake(
&database.account_identifier,
&token,
Some(column_order),
false,
)?
};
let r = run_future_with_polling_update_job_poller(
benchmarks/lib.ts
+1 -1
View File
@@ -2,7 +2,7 @@ import { sleep } from "https://deno.land/x/sleep@v1.2.1/mod.ts";
import * as windmill from "https://deno.land/x/windmill@v1.174.0/mod.ts";
import * as api from "https://deno.land/x/windmill@v1.174.0/windmill-api/index.ts";
export const VERSION = "v1.400.0";
export const VERSION = "v1.401.0";
export async function login(email: string, password: string): Promise<string> {
return await windmill.UserService.login({
cli/main.ts
+1 -1
View File
@@ -56,7 +56,7 @@ export {
// }
// });
export const VERSION = "1.400.0";
export const VERSION = "1.401.0";
const command = new Command()
.name("wmill")
frontend/package-lock.json
Generated
+2 -2
View File
@@ -1,12 +1,12 @@
{
"name": "windmill-components",
"version": "1.400.0",
"version": "1.401.0",
"lockfileVersion": 3,
"requires": true,
"packages": {
"": {
"name": "windmill-components",
"version": "1.400.0",
"version": "1.401.0",
"license": "AGPL-3.0",
"dependencies": {
"@aws-crypto/sha256-js": "^4.0.0",
frontend/package.json
+1 -1
View File
@@ -1,6 +1,6 @@
{
"name": "windmill-components",
"version": "1.400.0",
"version": "1.401.0",
"scripts": {
"dev": "vite dev",
"build": "vite build",
frontend/src/lib/components/graph/FlowGraphV2.svelte
+3 -1
View File
@@ -265,7 +265,9 @@
let yamlEditorDrawer: Drawer | undefined = undefined
</script>
<FlowYamlEditor bind:drawer={yamlEditorDrawer} />
{#if insertable}
<FlowYamlEditor bind:drawer={yamlEditorDrawer} />
{/if}
<div style={`height: ${height}px; max-height: ${maxHeight}px;`} bind:clientWidth={width}>
{#if graph?.error}
frontend/src/lib/script_helpers.ts
+2 -1
View File
@@ -215,7 +215,8 @@ INSERT INTO demo VALUES (?, ?);
UPDATE demo SET col2 = ? WHERE col2 = ?;
`
export const MSSQL_INIT_CODE = `-- to pin the database use '-- database f/your/path'
export const MSSQL_INIT_CODE = `-- return_last_result
-- to pin the database use '-- database f/your/path'
-- @p1 name1 (varchar) = default arg
-- @p2 name2 (int)
-- @p3 name3 (int)
lsp/Pipfile
+2 -2
View File
@@ -4,8 +4,8 @@ verify_ssl = true
name = "pypi"
[packages]
wmill = ">=1.400.0"
wmill_pg = ">=1.400.0"
wmill = ">=1.401.0"
wmill_pg = ">=1.401.0"
sendgrid = "*"
mysql-connector-python = "*"
pymongo = "*"
openflow.openapi.yaml
+1 -1
View File
@@ -1,7 +1,7 @@
openapi: "3.0.3"
info:
version: 1.400.0
version: 1.401.0
title: OpenFlow Spec
contact:
name: Ruben Fiszel
powershell-client/WindmillClient/WindmillClient.psd1
+1 -1
View File
@@ -12,7 +12,7 @@
RootModule = 'WindmillClient.psm1'
# Version number of this module.
ModuleVersion = '1.400.0'
ModuleVersion = '1.401.0'
# Supported PSEditions
# CompatiblePSEditions = @()
python-client/wmill/pyproject.toml
+1 -1
View File
@@ -1,6 +1,6 @@
[tool.poetry]
name = "wmill"
version = "1.400.0"
version = "1.401.0"
description = "A client library for accessing Windmill server wrapping the Windmill client API"
license = "Apache-2.0"
homepage = "https://windmill.dev"
python-client/wmill_pg/pyproject.toml
+1 -1
View File
@@ -1,6 +1,6 @@
[tool.poetry]
name = "wmill-pg"
version = "1.400.0"
version = "1.401.0"
description = "An extension client for the wmill client library focused on pg"
license = "Apache-2.0"
homepage = "https://windmill.dev"
typescript-client/jsr.json
+1 -1
View File
@@ -1,6 +1,6 @@
{
"name": "@windmill/windmill",
"version": "1.400.0",
"version": "1.401.0",
"exports": "./src/index.ts",
"publish": {
"exclude": ["!src", "./s3Types.ts", "./client.ts"]
typescript-client/package.json
+1 -1
View File
@@ -1,7 +1,7 @@
{
"name": "windmill-client",
"description": "Windmill SDK client for browsers and Node.js",
"version": "1.400.0",
"version": "1.401.0",
"author": "Ruben Fiszel",
"license": "Apache 2.0",
"devDependencies": {
version.txt
+1 -1
View File
@@ -1 +1 @@
1.400.0
1.401.0