feat: add focused draft cli edits

This commit is contained in:
lda
2026-06-15 18:27:27 +07:00 Verified
parent 577fb4707a
commit 97076479f1
20 changed files with 1832 additions and 6 deletions
+55
View File
@@ -98,6 +98,61 @@ def save_deployment(
] = None,
) -> None:
"""Save a workflow deployment from flags or a JSON object."""
_save_deployment_command(
ctx,
deployment_id=deployment_id,
artifact_id=artifact_id,
version=version,
binding=binding,
input_json=input_json,
input_file=input_file,
)
@app.command("create")
def create_deployment(
ctx: typer.Context,
deployment_id: Annotated[str | None, typer.Argument(help="Deployment id.")] = None,
artifact_id: Annotated[
str | None, typer.Option("--artifact", help="Artifact id.")
] = None,
version: Annotated[
int | None, typer.Option("--version", min=1, help="Artifact version.")
] = None,
binding: Annotated[
list[str] | None,
typer.Option("--binding", help="Logical=concrete source binding. Repeatable."),
] = None,
input_json: Annotated[
str | None, typer.Option("--input", help="Full deployment JSON object.")
] = None,
input_file: Annotated[
Path | None,
typer.Option("--input-file", help="Path to full deployment JSON object."),
] = None,
) -> None:
"""Alias for `deploy save`; creates or updates a deployment record."""
_save_deployment_command(
ctx,
deployment_id=deployment_id,
artifact_id=artifact_id,
version=version,
binding=binding,
input_json=input_json,
input_file=input_file,
)
def _save_deployment_command(
ctx: typer.Context,
*,
deployment_id: str | None,
artifact_id: str | None,
version: int | None,
binding: list[str] | None,
input_json: str | None,
input_file: Path | None,
) -> None:
try:
if input_json is not None or input_file is not None:
payload = parse_json_input(input_json=input_json, input_file=input_file)
+128
View File
@@ -10,6 +10,18 @@ from wf_cli.formats import ListOutputFormat, emit_list_payload
from wf_cli.io import CliInputError, emit_json, parse_bindings, parse_json_value
from wf_cli.remote_errors import run_cli_operation
def _parse_map_flags(values: list[str] | None) -> dict[str, str]:
parsed: dict[str, str] = {}
for item in values or []:
source, separator, target = item.partition("=")
if separator != "=" or not source or not target:
raise typer.BadParameter("--map must use source=target")
if source in parsed:
raise typer.BadParameter(f"duplicate --map for {source!r}")
parsed[source] = target
return parsed
app = typer.Typer(
name="draft",
help="Create, inspect, patch, validate, and save draft workflows.",
@@ -118,6 +130,122 @@ def patch_draft(
)
@app.command("set-name")
def set_draft_name(
ctx: typer.Context,
workspace_id: Annotated[str, typer.Argument(help="Draft workspace id.")],
revision: Annotated[
int, typer.Option("--revision", min=1, help="Expected workspace revision.")
],
name: Annotated[str, typer.Option("--name", help="New draft workflow name.")],
) -> None:
"""Set the draft workflow name without writing JSON Patch manually."""
context = load_cli_context(ctx)
emit_json(
run_cli_operation(
context,
context.handlers.set_draft_name(
workspace_id=workspace_id,
revision=revision,
name=name,
),
)
)
@app.command("set-route")
def set_draft_route(
ctx: typer.Context,
workspace_id: Annotated[str, typer.Argument(help="Draft workspace id.")],
revision: Annotated[
int, typer.Option("--revision", min=1, help="Expected workspace revision.")
],
step_id: Annotated[str, typer.Option("--step", help="Draft step id.")],
outcome: Annotated[str, typer.Option("--outcome", help="Step outcome.")],
target: Annotated[
str, typer.Option("--to", help="Target step id or __end__.")
],
) -> None:
"""Set one route: steps.<step> outcome -> target."""
context = load_cli_context(ctx)
emit_json(
run_cli_operation(
context,
context.handlers.set_draft_route(
workspace_id=workspace_id,
revision=revision,
step_id=step_id,
outcome=outcome,
target=target,
),
)
)
@app.command("set-input")
def set_step_input_map(
ctx: typer.Context,
workspace_id: Annotated[str, typer.Argument(help="Draft workspace id.")],
revision: Annotated[
int, typer.Option("--revision", min=1, help="Expected workspace revision.")
],
step_id: Annotated[str, typer.Option("--step", help="Draft step id.")],
mapping: Annotated[
list[str] | None,
typer.Option(
"--map",
help="Input binding SOURCE=LOCAL_TARGET. Repeatable. Example: input.text=text",
),
] = None,
) -> None:
"""Replace one step's input map without writing JSON Patch manually."""
input_map = _parse_map_flags(mapping)
context = load_cli_context(ctx)
emit_json(
run_cli_operation(
context,
context.handlers.set_step_input_map(
workspace_id=workspace_id,
revision=revision,
step_id=step_id,
input_map=input_map,
),
)
)
@app.command("set-output")
def set_step_output_map(
ctx: typer.Context,
workspace_id: Annotated[str, typer.Argument(help="Draft workspace id.")],
revision: Annotated[
int, typer.Option("--revision", min=1, help="Expected workspace revision.")
],
step_id: Annotated[str, typer.Option("--step", help="Draft step id.")],
mapping: Annotated[
list[str] | None,
typer.Option(
"--map",
help="Output binding LOCAL_SOURCE=STATE_TARGET. Repeatable. Example: text=state.text",
),
] = None,
) -> None:
"""Replace one step's output map without writing JSON Patch manually."""
output_map = _parse_map_flags(mapping)
context = load_cli_context(ctx)
emit_json(
run_cli_operation(
context,
context.handlers.set_step_output_map(
workspace_id=workspace_id,
revision=revision,
step_id=step_id,
output_map=output_map,
),
)
)
@app.command("validate")
def validate_draft(
ctx: typer.Context,