kalugin66/windmill
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: remove res1 wrapping

Browse Source
This commit is contained in:
Ruben Fiszel
2022-08-05 19:14:00 +02:00
parent c62d708993
commit e76a9816ee
6 changed files with 39 additions and 71 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
backend/src/flows.rs
View File

@@ -444,14 +444,14 @@ mod tests {
value: Box::new(FlowValue { modules: vec![], failure_module: None }),
skip_failures: true,
},
stop_after_if_expr: Some("previous.res1.isEmpty()".to_string()),
stop_after_if_expr: Some("previous.isEmpty()".to_string()),
skip_if_stopped: None,
},
],
failure_module: Some(FlowModule {
input_transform: HashMap::new(),
value: FlowModuleValue::Flow { path: "test".to_string() },
stop_after_if_expr: Some("previous.res1.isEmpty()".to_string()),
stop_after_if_expr: Some("previous.isEmpty()".to_string()),
skip_if_stopped: None,
}),
};

7
backend/src/jobs.rs
View File

@@ -1271,7 +1271,7 @@ pub async fn add_completed_job_error<E: ToString + std::fmt::Debug>(
&queued_job,
false,
false,
Some(output_map.clone()),
serde_json::Value::Object(output_map.clone()),
format!("\n{}\n{}", logs, e.to_string()),
)
.await?;
@@ -1284,10 +1284,9 @@ pub async fn add_completed_job(
queued_job: &QueuedJob,
success: bool,
skipped: bool,
result: Option<Map<String, Value>>,
result: serde_json::Value,
logs: String,
) -> Result<Uuid, Error> {
let result_json = result.map(serde_json::Value::Object);
let job_id = queued_job.id.clone();
let duration_ms = (chrono::Utc::now() - queued_job.started_at.unwrap_or(queued_job.created_at))
.num_milliseconds() as i32;
@@ -1311,7 +1310,7 @@ pub async fn add_completed_job(
queued_job.script_hash.map(|x| x.0),
queued_job.script_path,
queued_job.args,
result_json,
result,
logs,
queued_job.raw_code,
queued_job.canceled,

45
backend/src/worker.rs
View File

@@ -198,7 +198,7 @@ pub async fn run_worker(
db,
&job,
false,
Some(m),
serde_json::Value::Object(m),
&metrics,
)
.await;
@@ -277,11 +277,12 @@ async fn handle_queued_job(
match job.job_kind {
JobKind::FlowPreview | JobKind::Flow => {
let args = match &job.args {
Some(serde_json::Value::Object(m)) => Some(m.to_owned()),
_ => None,
};
handle_flow(&job, db, args).await?;
handle_flow(
&job,
db,
job.args.clone().unwrap_or_else(|| serde_json::Value::Null),
)
.await?;
}
_ => {
let mut logs = "".to_string();
@@ -314,10 +315,9 @@ async fn handle_queued_job(
match execution {
Ok(r) => {
add_completed_job(db, &job, true, false, r.result.clone(), logs).await?;
add_completed_job(db, &job, true, false, r.clone(), logs).await?;
if job.is_flow_step {
update_flow_status_after_job_completion(db, &job, true, r.result, metrics)
.await?;
update_flow_status_after_job_completion(db, &job, true, r, metrics).await?;
}
}
Err(e) => {
@@ -328,7 +328,7 @@ async fn handle_queued_job(
db,
&job,
false,
Some(output_map),
serde_json::Value::Object(output_map),
metrics,
)
.await?;
@@ -350,10 +350,6 @@ async fn handle_queued_job(
Ok(())
}
struct JobResult {
result: Option<Map<String, Value>>,
}
async fn write_file(dir: &str, path: &str, content: &str) -> Result<File, Error> {
let path = format!("{}/{}", dir, path);
let mut file = File::create(&path).await?;
@@ -405,7 +401,7 @@ async fn handle_job(
base_url: &str,
disable_nuser: bool,
disable_nsjail: bool,
) -> Result<JobResult, Error> {
) -> Result<serde_json::Value, Error> {
tracing::info!(
worker = %worker_name,
job_id = %job.id,
@@ -444,14 +440,14 @@ async fn handle_job(
tokio::fs::remove_dir_all(job_dir).await?;
if status.is_ok() && status.as_ref().unwrap().success() {
let result = serde_json::from_str::<Map<String, Value>>(last_line).map_err(|e| {
let result = serde_json::from_str::<serde_json::Value>(last_line).map_err(|e| {
Error::ExecutionErr(format!(
"result {} is not parsable.\n err: {}",
last_line,
e.to_string()
))
})?;
Ok(JobResult { result: Some(result) })
Ok(result)
} else {
let err = match status {
Ok(_) => {
@@ -639,12 +635,6 @@ for k, v in kwargs.items():
kwargs[k] = None
{transforms}
res = inner_script.main(**kwargs)
if res is None:
res = {{}}
if isinstance(res, tuple):
res = {{f"res{{i+1}}": v for i, v in enumerate(res)}}
if not isinstance(res, dict):
res = {{ "res1": res }}
res_json = json.dumps(res, separators=(',', ':'), default=str).replace('\n', '')
print()
print("result:")
@@ -748,14 +738,7 @@ const args = await Deno.readTextFile("args.json")
async function run() {{
let res: any = await main(...args);
if (res == undefined) {{
res = {{}}
}}
if (typeof res !== 'object' || Array.isArray(res)) {{
res = {{ res1: res }}
}}
const res_json = JSON.stringify(res);
const res_json = JSON.stringify(res ?? null);
console.log();
console.log("result:");
console.log(res_json);

30
backend/src/worker_flow.rs
View File

@@ -49,7 +49,7 @@ pub async fn update_flow_status_after_job_completion(
db: &DB,
job: &QueuedJob,
success: bool,
result: Option<Map<String, Value>>,
result: serde_json::Value,
metrics: &worker::Metrics,
) -> error::Result<()> {
tracing::debug!("HANDLE FLOW: {job:?} {success} {result:?}");
@@ -167,9 +167,7 @@ pub async fn update_flow_status_after_job_completion(
.map_ok(|(v,)| v)
.try_collect::<Vec<serde_json::Value>>()
.await?;
let mut results_map = serde_json::Map::new();
results_map.insert("res1".to_string(), serde_json::json!(results));
Some(results_map)
serde_json::json!(results)
}
_ => result.clone(),
};
@@ -252,11 +250,7 @@ async fn skip_loop_failures<'c>(
.map_err(|e| Error::InternalErr(format!("error during retrieval of skip_loop_failures: {e}")))
}
async fn compute_stop_early(
expr: String,
result: Option<Map<String, Value>>,
) -> error::Result<bool> {
let result = serde_json::Value::Object(result.clone().unwrap_or_else(|| Map::new()));
async fn compute_stop_early(expr: String, result: serde_json::Value) -> error::Result<bool> {
match eval_timeout(expr, [("result".to_string(), result)].into(), None, vec![]).await? {
serde_json::Value::Bool(true) => Ok(true),
serde_json::Value::Bool(false) => Ok(false),
@@ -311,7 +305,7 @@ pub async fn get_step_of_flow_status(db: &DB, id: Uuid) -> error::Result<i32> {
#[instrument(level = "trace", skip_all)]
async fn transform_input(
flow_args: &Option<serde_json::Value>,
last_result: Option<Map<String, serde_json::Value>>,
last_result: serde_json::Value,
input_transform: &HashMap<String, InputTransform>,
workspace: &str,
token: &str,
@@ -333,8 +327,7 @@ async fn transform_input(
match val {
InputTransform::Static { value: _ } => (),
InputTransform::Javascript { expr } => {
let previous_result =
serde_json::Value::Object(last_result.clone().unwrap_or_else(|| Map::new()));
let previous_result = last_result.clone();
let flow_input = flow_args.clone().unwrap_or_else(|| json!({}));
let v = eval_timeout(
expr.to_string(),
@@ -365,7 +358,7 @@ async fn transform_input(
pub async fn handle_flow(
flow_job: &QueuedJob,
db: &sqlx::Pool<sqlx::Postgres>,
last_result: Option<Map<String, serde_json::Value>>,
last_result: serde_json::Value,
) -> anyhow::Result<()> {
let value = flow_job
.raw_flow
@@ -397,7 +390,7 @@ async fn push_next_flow_job(
flow: FlowValue,
schedule_path: Option<String>,
db: &sqlx::Pool<sqlx::Postgres>,
last_result: Option<Map<String, serde_json::Value>>,
last_result: serde_json::Value,
) -> anyhow::Result<()> {
let flow_status_json = flow_job.flow_status.as_ref().ok_or_else(|| {
Error::InternalErr(format!("not found status for flow job {:?}", flow_job.id))
@@ -468,9 +461,7 @@ async fn push_next_flow_job(
let itered = match iterator {
InputTransform::Static { value } => value.clone(),
InputTransform::Javascript { expr } => {
let result = serde_json::Value::Object(
last_result.clone().unwrap_or_else(|| Map::new()),
);
let result = last_result.clone();
eval_timeout(
expr.to_string(),
[("result".to_string(), result)].into(),
@@ -480,9 +471,8 @@ async fn push_next_flow_job(
.await?
}
};
let mut args = last_result.clone().unwrap_or_else(Map::new);
let mut args = Map::new();
args.insert("_iterator".to_string(), last_result.clone());
args.insert(
"_index".to_string(),
serde_json::Value::Number(serde_json::Number::from(0)),

14
frontend/src/lib/components/DisplayResult.svelte
View File

@@ -9,17 +9,13 @@
let resultKind: 'json' | 'table-col' | 'table-row' | 'png' | 'file' | 'jpeg' | 'gif' | undefined =
inferResultKind(result)
function isArray(obj: any) {
return Object.prototype.toString.call(obj) === '[object Array]'
}
function isRectangularArray(obj: any) {
if (!isArray(obj) || obj.length == 0) {
if (!Array.isArray(obj) || obj.length == 0) {
return false
}
if (
!Object.values(obj)
.map(isArray)
.map(Array.isArray)
.reduce((a, b) => a && b)
) {
return false
@@ -39,9 +35,9 @@
if (result) {
try {
let keys = Object.keys(result)
if (keys.length == 1 && isRectangularArray(result[keys[0]])) {
if (isRectangularArray(result)) {
return 'table-row'
} else if (keys.map((k) => isArray(result[k])).reduce((a, b) => a && b)) {
} else if (keys.map((k) => Array.isArray(result[k])).reduce((a, b) => a && b)) {
return 'table-col'
} else if (keys.length == 1 && keys[0] == 'png') {
return 'png'
@@ -57,7 +53,7 @@
</script>
{#if result}
{#if Object.keys(result).length > 0}<div>
{#if typeof result == 'object' && Object.keys(result).length > 0}<div>
The result keys are: <b>{Object.keys(result).join(', ')}</b>
</div>
{/if}

10
frontend/src/lib/components/flows/utils.ts
View File

@@ -36,7 +36,7 @@ export function flowToMode(flow: Flow | any, mode: FlowMode): Flow {
const oldModules = newFlow.value.modules.slice(1)
if (triggerModule) {
triggerModule.stop_after_if_expr = 'result.res1.length == 0'
triggerModule.stop_after_if_expr = 'result.length == 0'
triggerModule.skip_if_stopped = true
}
@@ -46,7 +46,7 @@ export function flowToMode(flow: Flow | any, mode: FlowMode): Flow {
input_transform: oldModules[0].input_transform,
value: {
type: 'forloopflow',
iterator: { type: 'javascript', expr: 'result.res1' },
iterator: { type: 'javascript', expr: 'result' },
value: {
modules: oldModules
},
@@ -231,10 +231,10 @@ export async function runFlowPreview(args: Record<string, any>, flow: Flow) {
})
}
function computeFlowInputPull(previewResult: any | undefined, flowInputAsObject: any) {
const iteratorValues = (previewResult?.res1 && Array.isArray(previewResult.res1)) ?
const iteratorValues = (previewResult && Array.isArray(previewResult)) ?
{
_value: previewResult.res1[0],
_index: `The current index of the iteration as a number (here from 0 to ${previewResult.res1.length - 1})`
_value: previewResult[0],
_index: `The current index of the iteration as a number (here from 0 to ${previewResult.length - 1})`
} : {
_value: 'The current value of the iteration as an object',
_index: 'The current index of the iteration as a number'