diff --git a/backend/windmill-worker/src/python_executor.rs b/backend/windmill-worker/src/python_executor.rs index 90be4e9cfc..ad46bfd3cd 100644 --- a/backend/windmill-worker/src/python_executor.rs +++ b/backend/windmill-worker/src/python_executor.rs @@ -66,10 +66,10 @@ lazy_static::lazy_static! { lazy_static::lazy_static! { static ref PIPTAR_UPLOAD_CHANNEL: tokio::sync::mpsc::UnboundedSender = { let (tx, rx) = tokio::sync::mpsc::unbounded_channel(); - + // Spawn background task to handle uploads sequentially tokio::spawn(handle_piptar_uploads(rx)); - + tx }; } @@ -85,7 +85,7 @@ struct PiptarUploadTask { async fn handle_piptar_uploads(mut rx: tokio::sync::mpsc::UnboundedReceiver) { use crate::global_cache::build_tar_and_push; use windmill_common::s3_helpers::get_object_store; - + while let Some(task) = rx.recv().await { if let Some(os) = get_object_store().await { match build_tar_and_push(os, task.venv_path.clone(), task.cache_dir, None, false).await { @@ -324,6 +324,7 @@ pub async fn uv_pip_compile( child_cmd .env("SystemRoot", SYSTEM_ROOT.as_str()) .env("USERPROFILE", crate::USERPROFILE_ENV.as_str()) + .env("HOME", crate::USERPROFILE_ENV.as_str()) .env( "LOCALAPPDATA", std::env::var("LOCALAPPDATA") @@ -332,6 +333,29 @@ pub async fn uv_pip_compile( .env( "TMP", std::env::var("TMP").unwrap_or_else(|_| String::from("/tmp")), + ) + .env( + "APPDATA", + std::env::var("APPDATA") + .unwrap_or_else(|_| format!("{}\\AppData\\Roaming", crate::USERPROFILE_ENV.as_str())), + ) + .env( + "ComSpec", + std::env::var("ComSpec").unwrap_or_else(|_| String::from("C:\\Windows\\System32\\cmd.exe")), + ) + .env( + "PATHEXT", + std::env::var("PATHEXT").unwrap_or_else(|_| + String::from(".COM;.EXE;.BAT;.CMD;.VBS;.VBE;.JS;.JSE;.WSF;.WSH;.MSC;.CPL") + ), + ) + .env( + "ProgramData", + std::env::var("ProgramData").unwrap_or_else(|_| String::from("C:\\ProgramData")), + ) + .env( + "ProgramFiles", + std::env::var("ProgramFiles").unwrap_or_else(|_| String::from("C:\\Program Files")), ); } @@ -602,7 +626,7 @@ pub async fn handle_python_job( if v == '': del pre_args[k] kwargs = inner_script.preprocessor(**pre_args) - kwrags_json = res_to_json(kwargs) + kwrags_json = res_to_json(kwargs) with open("args.json", 'w') as f: f.write(kwrags_json)"# ) @@ -678,7 +702,7 @@ except BaseException as e: tb = traceback.format_tb(exc_traceback) with open(result_json, 'w') as f: err = {{ "message": str(e), "name": e.__class__.__name__, "stack": '\n'.join(tb[1:]) }} - extra = e.__dict__ + extra = e.__dict__ if extra and len(extra) > 0: err['extra'] = extra flow_node_id = os.environ.get('WM_FLOW_STEP_ID') @@ -1417,6 +1441,7 @@ async fn spawn_uv_install( .envs(PROXY_ENVS.clone()) .env("SystemRoot", SYSTEM_ROOT.as_str()) .env("USERPROFILE", crate::USERPROFILE_ENV.as_str()) + .env("HOME", HOME_ENV.as_str()) .env( "TMP", std::env::var("TMP").unwrap_or_else(|_| String::from("/tmp")), @@ -1426,6 +1451,29 @@ async fn spawn_uv_install( std::env::var("LOCALAPPDATA") .unwrap_or_else(|_| format!("{}\\AppData\\Local", HOME_ENV.as_str())), ) + .env( + "APPDATA", + std::env::var("APPDATA") + .unwrap_or_else(|_| format!("{}\\AppData\\Roaming", crate::USERPROFILE_ENV.as_str())), + ) + .env( + "ComSpec", + std::env::var("ComSpec").unwrap_or_else(|_| String::from("C:\\Windows\\System32\\cmd.exe")), + ) + .env( + "PATHEXT", + std::env::var("PATHEXT").unwrap_or_else(|_| + String::from(".COM;.EXE;.BAT;.CMD;.VBS;.VBE;.JS;.JSE;.WSF;.WSH;.MSC;.CPL") + ), + ) + .env( + "ProgramData", + std::env::var("ProgramData").unwrap_or_else(|_| String::from("C:\\ProgramData")), + ) + .env( + "ProgramFiles", + std::env::var("ProgramFiles").unwrap_or_else(|_| String::from("C:\\Program Files")), + ) .args(&command_args[1..]) .stdout(Stdio::piped()) .stderr(Stdio::piped()); @@ -1810,7 +1858,7 @@ pub async fn handle_python_reqs( // Create a file to indicate that installation was successfull let valid_path = venv_p.clone() + "/.valid.windmill"; - // This is atomic operation, meaning, that it either completes and wheel is valid, + // This is atomic operation, meaning, that it either completes and wheel is valid, // or it does not and wheel is invalid and will be reinstalled next run if let Err(e) = File::create(&valid_path).await{ tracing::error!( @@ -1942,7 +1990,7 @@ pub async fn handle_python_reqs( venv_path: venv_p.clone(), cache_dir: py_version.to_cache_dir_top_level(false), }; - + if let Err(e) = PIPTAR_UPLOAD_CHANNEL.send(upload_task) { tracing::warn!("Failed to queue piptar upload for {venv_p}: {e}"); } else { @@ -1961,7 +2009,7 @@ pub async fn handle_python_reqs( pids.lock().await.get_mut(i).and_then(|e| e.take()); // Create a file to indicate that installation was successfull let valid_path = venv_p.clone() + "/.valid.windmill"; - // This is atomic operation, meaning, that it either completes and wheel is valid, + // This is atomic operation, meaning, that it either completes and wheel is valid, // or it does not and wheel is invalid and will be reinstalled next run if let Err(e) = File::create(&valid_path).await{ tracing::error!(