script gen run code block using the block interface (#3401)

This commit is contained in:
Shuchang Zheng
2025-09-09 22:33:59 -07:00
committed by GitHub
parent 53c62bf5fb
commit a903170f14
3 changed files with 99 additions and 15 deletions

View File

@@ -127,7 +127,9 @@ def _value(value: Any) -> cst.BaseExpression:
"""Convert simple Python objects to CST expressions."""
if isinstance(value, str):
if "\n" in value:
return cst.SimpleString('"""' + value.replace('"""', '\\"\\"\\"') + '"""')
# For multi-line strings, use repr() which handles all escaping properly
# This will use triple quotes when appropriate and escape them when needed
return cst.SimpleString(repr(value))
return cst.SimpleString(repr(value))
if isinstance(value, (int, float, bool)) or value is None:
return cst.parse_expression(repr(value))
@@ -877,6 +879,47 @@ def _build_goto_statement(block: dict[str, Any]) -> cst.SimpleStatementLine:
return cst.SimpleStatementLine([cst.Expr(cst.Await(call))])
def _build_code_statement(block: dict[str, Any]) -> cst.SimpleStatementLine:
"""Build a skyvern.run_code statement."""
args = [
cst.Arg(
keyword=cst.Name("code"),
value=_value(block.get("code", "")),
whitespace_after_arg=cst.ParenthesizedWhitespace(
indent=True,
last_line=cst.SimpleWhitespace(INDENT),
),
),
cst.Arg(
keyword=cst.Name("label"),
value=_value(block.get("label") or block.get("title") or f"block_{block.get('workflow_run_block_id')}"),
whitespace_after_arg=cst.ParenthesizedWhitespace(
indent=True,
last_line=cst.SimpleWhitespace(INDENT),
),
),
cst.Arg(
keyword=cst.Name("parameters"),
value=_value(block.get("parameters", None)),
whitespace_after_arg=cst.ParenthesizedWhitespace(
indent=True,
),
comma=cst.Comma(),
),
]
call = cst.Call(
func=cst.Attribute(value=cst.Name("skyvern"), attr=cst.Name("run_code")),
args=args,
whitespace_before_args=cst.ParenthesizedWhitespace(
indent=True,
last_line=cst.SimpleWhitespace(INDENT),
),
)
return cst.SimpleStatementLine([cst.Expr(cst.Await(call))])
def __build_base_task_statement(block_title: str, block: dict[str, Any]) -> list[cst.Arg]:
args = [
cst.Arg(
@@ -986,6 +1029,8 @@ def _build_run_fn(blocks: list[dict[str, Any]], wf_req: dict[str, Any]) -> Funct
stmt = _build_for_loop_statement(block_title, block)
elif block_type == "goto_url":
stmt = _build_goto_statement(block)
elif block_type == "code":
stmt = _build_code_statement(block)
else:
# Default case for unknown block types
stmt = cst.SimpleStatementLine([cst.Expr(cst.SimpleString(f"# Unknown block type: {block_type}"))])