lda.chat/lda-wf
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: add capability step update cli

Browse Source
This commit is contained in:
lda
2026-07-27 02:06:52 +07:00 Verified
parent 8d731444eb
commit bf8be6b3dc
7 changed files with 689 additions and 17 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/wf_cli/commands/draft_add.py
+45 -2
View File
@@ -40,7 +40,10 @@ from .draft_options import (
_parse_output_map_flags,
_parse_route_flags,
_parse_step_input_map_flags,
parse_capability_input_binding_flags,
parse_json_file,
parse_step_input_bindings_file,
parse_step_input_value_flags,
route_source,
validation_error_as_bad_parameter,
)
@@ -123,6 +126,20 @@ def add_step_from_capability(
),
),
] = None,
input_value: Annotated[
list[str] | None,
typer.Option(
"--value",
help="Literal input binding LOCAL_TARGET=JSON. Repeat as needed.",
),
] = None,
bindings_file: Annotated[
Path | None,
typer.Option(
"--bindings-file",
help="Ordered canonical JSON input-binding list.",
),
] = None,
output_mapping: Annotated[
list[str] | None,
typer.Option(
@@ -134,6 +151,16 @@ def add_step_from_capability(
),
),
] = None,
description: Annotated[
str | None, typer.Option("--description", help="Step description.")
] = None,
retry: Annotated[
int | None, typer.Option("--retry", min=0, help="Retry count.")
] = None,
timeout_seconds: Annotated[
int | None,
typer.Option("--timeout-seconds", min=1, help="Timeout in seconds."),
] = None,
) -> None:
"""Add a capability step; this also projects its schemas and bindings.
@@ -149,7 +176,19 @@ def add_step_from_capability(
`--input state.title=report.title --input state.summary=report.summary`
`--bind-output title=state.title --bind-output summary=state.summary`
"""
input_map = _parse_step_input_map_flags(input_mapping, option_name="--input")
convenience_input_selected = input_mapping is not None or input_value is not None
if bindings_file is not None and convenience_input_selected:
raise typer.BadParameter(
"--bindings-file is mutually exclusive with --input and --value"
)
input_bindings = (
parse_step_input_bindings_file(bindings_file)
if bindings_file is not None
else [
*parse_capability_input_binding_flags(input_mapping),
*parse_step_input_value_flags(input_value),
]
)
bind_outputs = _parse_output_map_flags(output_mapping)
routes = _parse_route_flags(route)
context = load_cli_context(ctx)
@@ -164,8 +203,12 @@ def add_step_from_capability(
route_from_step=route_from_step,
route_from_outcome=route_from_outcome,
routes=routes or None,
input_map=input_map,
input_map=None,
input_bindings=input_bindings,
bind_outputs=bind_outputs,
desc=description,
retry=retry,
timeout_seconds=timeout_seconds,
),
)
)
src/wf_cli/commands/draft_options.py
+28 -6
View File
@@ -108,12 +108,28 @@ def parse_step_input_binding_flags(
values: list[str] | None,
) -> list[InputPathBinding]:
"""Parse ordered step-input path bindings without collapsing source fan-out."""
return _parse_input_path_binding_flags(values, target_label="node-local")
return _parse_input_path_binding_flags(
values,
option_name="--map",
target_label="node-local",
)
def parse_capability_input_binding_flags(
values: list[str] | None,
) -> list[InputPathBinding]:
"""Parse ordered capability input paths from the public ``--input`` flag."""
return _parse_input_path_binding_flags(
values,
option_name="--input",
target_label="node-local",
)
def _parse_input_path_binding_flags(
values: list[str] | None,
*,
option_name: str,
target_label: str,
) -> list[InputPathBinding]:
"""Parse ordered GRAPH_SOURCE=LOCAL_TARGET bindings for one CLI audience."""
@@ -121,24 +137,26 @@ def _parse_input_path_binding_flags(
for item in values or []:
source, separator, target = item.partition("=")
if separator != "=" or not source or not target:
raise typer.BadParameter("--map must use GRAPH_SOURCE=LOCAL_TARGET")
raise typer.BadParameter(
f"{option_name} must use GRAPH_SOURCE=LOCAL_TARGET"
)
if target.startswith("local."):
bare_target = target.removeprefix("local.")
raise typer.BadParameter(
f"--map target must be a rootless {target_label} path; "
f"{option_name} target must be a rootless {target_label} path; "
f"use {source}={bare_target}, not {source}={target}"
)
try:
source_path = GraphSourcePath.parse(source)
except PathResolutionError as exc:
raise typer.BadParameter(
f"--map source must be a graph source path: {exc}"
f"{option_name} source must be a graph source path: {exc}"
) from exc
try:
target_path = LocalPath.parse(target)
except PathResolutionError as exc:
raise typer.BadParameter(
f"--map target must be a rootless {target_label} path: {exc}"
f"{option_name} target must be a rootless {target_label} path: {exc}"
) from exc
bindings.append(InputPathBinding(path=source_path, target=target_path))
return bindings
@@ -198,7 +216,11 @@ def parse_workflow_output_binding_flags(
values: list[str] | None,
) -> list[InputPathBinding]:
"""Parse ordered canonical workflow-output path bindings."""
return _parse_input_path_binding_flags(values, target_label="workflow-output")
return _parse_input_path_binding_flags(
values,
option_name="--map",
target_label="workflow-output",
)
def parse_workflow_output_value_flags(
src/wf_cli/commands/draft_update.py
+170
View File
@@ -0,0 +1,170 @@
from __future__ import annotations
from pathlib import Path
from typing import Annotated
import typer
from pydantic import ValidationError
from wf_api import CapabilityStepUpdate
from wf_cli.context import load_cli_context_from_typer as load_cli_context
from wf_cli.io import emit_json
from wf_cli.remote_errors import run_cli_operation
from .draft_options import (
parse_capability_input_binding_flags,
parse_step_input_bindings_file,
parse_step_input_value_flags,
validation_error_as_bad_parameter,
)
app = typer.Typer(
name="update",
help="Update one existing typed draft step.",
no_args_is_help=True,
)
def _reject_set_clear_conflict(
*,
value_is_set: bool,
clear: bool,
value_option: str,
clear_option: str,
) -> None:
if value_is_set and clear:
raise typer.BadParameter(
f"{value_option} and {clear_option} are mutually exclusive"
)
@app.command("capability")
def update_capability_step(
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="Existing draft step id.")],
description: Annotated[
str | None, typer.Option("--description", help="Replace the step description.")
] = None,
clear_description: Annotated[
bool,
typer.Option("--clear-description", help="Remove the step description."),
] = False,
retry: Annotated[
int | None, typer.Option("--retry", min=0, help="Replace the retry count.")
] = None,
clear_retry: Annotated[
bool, typer.Option("--clear-retry", help="Remove the retry override.")
] = False,
timeout_seconds: Annotated[
int | None,
typer.Option(
"--timeout-seconds",
min=1,
help="Replace the timeout in seconds.",
),
] = None,
clear_timeout: Annotated[
bool, typer.Option("--clear-timeout", help="Remove the timeout override.")
] = False,
input_mapping: Annotated[
list[str] | None,
typer.Option(
"--input",
help="Input binding GRAPH_SOURCE=LOCAL_TARGET. Repeat as needed.",
),
] = None,
input_value: Annotated[
list[str] | None,
typer.Option(
"--value",
help="Literal input binding LOCAL_TARGET=JSON. Repeat as needed.",
),
] = None,
bindings_file: Annotated[
Path | None,
typer.Option(
"--bindings-file",
help="Ordered canonical JSON input-binding list.",
),
] = None,
clear_input: Annotated[
bool,
typer.Option("--clear-input", help="Replace input with no bindings."),
] = False,
) -> None:
"""Patch metadata or replace all input bindings on a capability step."""
_reject_set_clear_conflict(
value_is_set=description is not None,
clear=clear_description,
value_option="--description",
clear_option="--clear-description",
)
_reject_set_clear_conflict(
value_is_set=retry is not None,
clear=clear_retry,
value_option="--retry",
clear_option="--clear-retry",
)
_reject_set_clear_conflict(
value_is_set=timeout_seconds is not None,
clear=clear_timeout,
value_option="--timeout-seconds",
clear_option="--clear-timeout",
)
convenience_input_selected = input_mapping is not None or input_value is not None
if bindings_file is not None and (convenience_input_selected or clear_input):
raise typer.BadParameter(
"--bindings-file is mutually exclusive with --input, --value, "
"and --clear-input"
)
if clear_input and convenience_input_selected:
raise typer.BadParameter(
"--clear-input is mutually exclusive with --input and --value"
)
payload: dict[str, object] = {}
if description is not None:
payload["desc"] = description
elif clear_description:
payload["desc"] = None
if retry is not None:
payload["retry"] = retry
elif clear_retry:
payload["retry"] = None
if timeout_seconds is not None:
payload["timeout_seconds"] = timeout_seconds
elif clear_timeout:
payload["timeout_seconds"] = None
if bindings_file is not None:
payload["input"] = parse_step_input_bindings_file(bindings_file)
elif clear_input:
payload["input"] = []
elif convenience_input_selected:
payload["input"] = [
*parse_capability_input_binding_flags(input_mapping),
*parse_step_input_value_flags(input_value),
]
try:
update = CapabilityStepUpdate.model_validate(payload)
except ValidationError as exc:
raise validation_error_as_bad_parameter(exc) from exc
context = load_cli_context(ctx)
emit_json(
run_cli_operation(
context,
context.handlers.update_capability_step(
workspace_id=workspace_id,
revision=revision,
step_id=step_id,
update=update,
),
)
)
src/wf_cli/commands/drafts.py
+2 -1
View File
@@ -6,7 +6,7 @@ from typing import Annotated, Literal
import typer
from wf_cli.commands import draft_add
from wf_cli.commands import draft_add, draft_update
from wf_cli.commands.draft_options import (
_parse_map_flags,
_parse_output_map_flags,
@@ -33,6 +33,7 @@ app = typer.Typer(
no_args_is_help=True,
)
app.add_typer(draft_add.app, name="add")
app.add_typer(draft_update.app, name="update")
def _validate_outcomes(values: list[str] | None) -> tuple[str, ...] | None: