diff --git a/backend/src/flows.rs b/backend/src/flows.rs index 20a81b901b..05de561abc 100644 --- a/backend/src/flows.rs +++ b/backend/src/flows.rs @@ -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, }), }; diff --git a/backend/src/jobs.rs b/backend/src/jobs.rs index b9bd7deaff..9e91993a5b 100644 --- a/backend/src/jobs.rs +++ b/backend/src/jobs.rs @@ -1271,7 +1271,7 @@ pub async fn add_completed_job_error( &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>, + result: serde_json::Value, logs: String, ) -> Result { - 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, diff --git a/backend/src/worker.rs b/backend/src/worker.rs index 17f72aacdd..0b85262c92 100644 --- a/backend/src/worker.rs +++ b/backend/src/worker.rs @@ -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>, -} - async fn write_file(dir: &str, path: &str, content: &str) -> Result { 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 { +) -> Result { 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::>(last_line).map_err(|e| { + let result = serde_json::from_str::(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); diff --git a/backend/src/worker_flow.rs b/backend/src/worker_flow.rs index 4a6597ecef..795f97b0a7 100644 --- a/backend/src/worker_flow.rs +++ b/backend/src/worker_flow.rs @@ -49,7 +49,7 @@ pub async fn update_flow_status_after_job_completion( db: &DB, job: &QueuedJob, success: bool, - result: Option>, + 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::>() .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>, -) -> error::Result { - 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 { 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 { #[instrument(level = "trace", skip_all)] async fn transform_input( flow_args: &Option, - last_result: Option>, + last_result: serde_json::Value, input_transform: &HashMap, 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, - last_result: Option>, + 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, db: &sqlx::Pool, - last_result: Option>, + 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)), diff --git a/frontend/src/lib/components/DisplayResult.svelte b/frontend/src/lib/components/DisplayResult.svelte index e81c88206b..5dbf04004e 100644 --- a/frontend/src/lib/components/DisplayResult.svelte +++ b/frontend/src/lib/components/DisplayResult.svelte @@ -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 @@ {#if result} - {#if Object.keys(result).length > 0}
+ {#if typeof result == 'object' && Object.keys(result).length > 0}
The result keys are: {Object.keys(result).join(', ')}
{/if} diff --git a/frontend/src/lib/components/flows/utils.ts b/frontend/src/lib/components/flows/utils.ts index 2e120365d3..d08994ce55 100644 --- a/frontend/src/lib/components/flows/utils.ts +++ b/frontend/src/lib/components/flows/utils.ts @@ -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, 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'