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

feat: add preprocessor support for dedicated workers and bunnative scripts (#8284)

Browse Source 
* feat: add preprocessor support for dedicated workers

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

* chore: update ee-repo-ref.txt

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

* refactor: extract transform_and_run helper in python dedicated wrapper

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

* feat: add preprocessor support for bunnative scripts

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>

* fix: resolve unused postprocessor variable in python wrapper

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>

* chore: add workflow_dispatch trigger to backend integration tests

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>

* fix: bunnative fixture lock format and PrewarmedIsolate::spawn callers

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>

* fix: update generate_dedicated_worker_wrapper callers in bun_jobs test

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>

* fix: use non-dedicated workers in preprocessor integration tests

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>

* fix: prewarm preprocessor isolate for bunnative dedicated workers

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

* refactor: flatten bunnative dedicated worker preprocessing into single result path

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

* fix: use labeled block instead of async block for EE compatibility

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

* Update commit reference in ee-repo-ref.txt

* chore: update ee-repo-ref to e36945b987f7904fa984181baf3124e7b2722bd1

This commit updates the EE repository reference after PR #445 was merged in windmill-ee-private.

Previous ee-repo-ref: 8a2625833452aadb8907242bf502b24ca2dffd73

New ee-repo-ref: e36945b987f7904fa984181baf3124e7b2722bd1

Automated by sync-ee-ref workflow.

* Fix merge conflict in ee-repo-ref.txt

Resolve merge conflict in ee-repo-ref.txt

---------

Co-authored-by: Claude Opus 4.6 <noreply@anthropic.com>
Co-authored-by: windmill-internal-app[bot] <windmill-internal-app[bot]@users.noreply.github.com>
This commit is contained in:
hugocasa
2026-03-10 18:26:02 +00:00
committed by GitHub
co-authored by Claude Opus 4.6 windmill-internal-app[bot]
parent fefc8c62a0
commit dc0e59f432
9 changed files with 784 additions and 169 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/windmill-worker/src/python_executor.rs
+109 -60
View File
@@ -680,6 +680,7 @@ pub async fn handle_python_job(
String::new()
};
let main_override = main_name.unwrap_or_else(|| "main".to_string());
let res_to_json_body = python_res_to_json_body(postprocessor);
let wrapper_content: String = if is_wac_v2 {
format!(
r#"
@@ -768,19 +769,7 @@ replace_invalid_fields = re.compile(r'(?:\bNaN\b|\\*\\u0000|Infinity|\-Infinity)
result_json = os.path.join(os.path.abspath(os.path.dirname(__file__)), "result.json")
def res_to_json(res, typ):
if typ.__name__ == 'DataFrame':
if typ.__module__ == 'pandas.core.frame':
res = res.values.tolist()
elif typ.__module__ == 'polars.dataframe.frame':
res = res.rows()
elif typ.__name__ == 'bytes':
res = to_b_64(res)
elif typ.__name__ == 'dict':
for k, v in res.items():
if type(v).__name__ == 'bytes':
res[k] = to_b_64(v)
unprocessed = json.dumps(res, separators=(',', ':'), default=str).replace('\n', '')
return {postprocessor}
{res_to_json_body}
try:
{preprocessor}
@@ -1027,6 +1016,30 @@ mount {{
Ok(result)
}
/// Generate Python code to spread preprocessor args from `kwargs` into `pre_args`.
/// The `indent` parameter controls the join separator for multi-arg spreads.
fn python_preprocessor_spread(sig: windmill_parser::MainArgSignature, indent: &str) -> String {
if sig.star_kwargs {
"pre_args = kwargs".to_string()
} else {
sig.args
.into_iter()
.map(|x| {
let name = &x.name;
if x.default.is_none() {
format!("pre_args[\"{name}\"] = kwargs.get(\"{name}\")")
} else {
format!(
r#"pre_args["{name}"] = kwargs.get("{name}")
{indent}if pre_args["{name}"] is None:
{indent} del pre_args["{name}"]"#
)
}
})
.join(&format!("\n{indent}"))
}
}
async fn prepare_wrapper(
job_dir: &str,
job_flow_step_id: Option<&str>,
@@ -1193,31 +1206,7 @@ async fn prepare_wrapper(
.join("\n ")
};
let pre_spread = if let Some(pre_sig) = pre_sig {
let spread = if pre_sig.star_kwargs {
"pre_args = kwargs".to_string()
} else {
pre_sig
.args
.into_iter()
.map(|x| {
let name = &x.name;
if x.default.is_none() {
format!("pre_args[\"{name}\"] = kwargs.get(\"{name}\")")
} else {
format!(
r#"pre_args["{name}"] = kwargs.get("{name}")
if pre_args["{name}"] is None:
del pre_args["{name}"]"#
)
}
})
.join("\n ")
};
Some(spread)
} else {
None
};
let pre_spread = pre_sig.map(|sig| python_preprocessor_spread(sig, " "));
let module_dir_dot = dirs.replace("/", ".").replace("-", "_");
@@ -2338,6 +2327,26 @@ This is not normal behavior, please make sure all workers have enough memory.\n
};
}
/// Python function body for `res_to_json(res, typ)`.
/// Handles DataFrame, bytes, dict coercion + JSON serialization.
fn python_res_to_json_body(postprocessor: &str) -> String {
format!(
r#" if typ.__name__ == 'DataFrame':
if typ.__module__ == 'pandas.core.frame':
res = res.values.tolist()
elif typ.__module__ == 'polars.dataframe.frame':
res = res.rows()
elif typ.__name__ == 'bytes':
res = to_b_64(res)
elif typ.__name__ == 'dict':
for k, v in res.items():
if type(v).__name__ == 'bytes':
res[k] = to_b_64(v)
unprocessed = json.dumps(res, separators=(',', ':'), default=str).replace('\n', '')
return {postprocessor}"#
)
}
// Returns code snippet that needs to be injected into wrapper to post-process results or leave unprocessed
fn get_result_postprocessor<'a>(skip: bool) -> &'a str {
if skip {
@@ -2433,6 +2442,16 @@ pub async fn start_worker(
_,
) = prepare_wrapper(job_dir, None, None, None, inner_content, script_path).await?;
// Parse preprocessor signature if the script has one
let pre_spread = windmill_parser_py::parse_python_signature(
inner_content,
Some("preprocessor".to_string()),
false,
)
.ok()
.filter(|sig| !sig.args.is_empty())
.map(|sig| python_preprocessor_spread(sig, " "));
{
let postprocessor = get_result_postprocessor(annotations.skip_result_postprocessing);
let indented_transforms = transforms
@@ -2441,6 +2460,41 @@ pub async fn start_worker(
.collect::<Vec<String>>()
.join("\n");
let preprocessor_logic = if let Some(ref pre_spread) = pre_spread {
format!(
r#"
if line.startswith('preprocess:'):
pre_input = line[len('preprocess:'):]
kwargs = json.loads(pre_input, strict=False)
if not hasattr(inner_script, 'preprocessor') or not callable(inner_script.preprocessor):
err_json = json.dumps({{"message": "preprocessor function is missing", "name": "Error"}}, separators=(',', ':'), default=str).replace('\n', '')
sys.stdout.write("wm_res[error]:" + err_json + "\n")
sys.stdout.flush()
continue
try:
pre_args = {{}}
{pre_spread}
for k, v in list(pre_args.items()):
if v == '<function call>':
del pre_args[k]
preprocessed_kwargs = inner_script.preprocessor(**pre_args)
preprocessed_json = json.dumps(preprocessed_kwargs, separators=(',', ':'), default=str).replace('\n', '')
sys.stdout.write("wm_res[preprocessed_args]:" + preprocessed_json + "\n")
transform_and_run(preprocessed_kwargs)
except BaseException as e:
exc_type, exc_value, exc_traceback = sys.exc_info()
tb = traceback.format_tb(exc_traceback)
err_json = json.dumps({{ "message": str(e), "name": e.__class__.__name__, "stack": '\n'.join(tb[1:]) }}, separators=(',', ':'), default=str).replace('\n', '')
sys.stdout.write("wm_res[error]:" + err_json + "\n")
sys.stdout.flush()
continue
"#
)
} else {
String::new()
};
let res_to_json_body = python_res_to_json_body(postprocessor);
let wrapper_content: String = format!(
r#"
import json
@@ -2458,37 +2512,32 @@ def to_b_64(v: bytes):
b64 = base64.b64encode(v)
return b64.decode('ascii')
replace_invalid_fields = re.compile(r'(?:\bNaN\b|\\u0000|Infinity|\-Infinity)')
sys.stdout.write('start\n')
def res_to_json(res, typ):
{res_to_json_body}
for line in sys.stdin:
if line == 'end\n':
break
kwargs = json.loads(line, strict=False)
def transform_and_run(kwargs):
args = {{}}
{indented_transforms}
{spread}
for k, v in list(args.items()):
if v == '<function call>':
del args[k]
res = inner_script.main(**args)
typ = type(res)
res_json = res_to_json(res, typ)
sys.stdout.write("wm_res[success]:" + res_json + "\n")
replace_invalid_fields = re.compile(r'(?:\bNaN\b|\\u0000|Infinity|\-Infinity)')
sys.stdout.write('start\n')
for line in sys.stdin:
if line == 'end\n':
break
line = line.strip()
{preprocessor_logic}
kwargs = json.loads(line, strict=False)
try:
res = inner_script.main(**args)
typ = type(res)
if typ.__name__ == 'DataFrame':
if typ.__module__ == 'pandas.core.frame':
res = res.values.tolist()
elif typ.__module__ == 'polars.dataframe.frame':
res = res.rows()
elif typ.__name__ == 'bytes':
res = to_b_64(res)
elif typ.__name__ == 'dict':
for k, v in res.items():
if type(v).__name__ == 'bytes':
res[k] = to_b_64(v)
unprocessed = json.dumps(res, separators=(',', ':'), default=str).replace('\n', '')
res_json = {postprocessor}
sys.stdout.write("wm_res[success]:" + res_json + "\n")
transform_and_run(kwargs)
except BaseException as e:
exc_type, exc_value, exc_traceback = sys.exc_info()
tb = traceback.format_tb(exc_traceback)