Skip to content

workflow_cmd

workflow_cmd

diapason workflow — workflow management commands.

Functions:

workflow 

workflow() -> None

Manage workflows — list, run, status.

Source code in src/diapason/cli/workflow_cmd.py 
10
11
12
@click.group()
def workflow() -> None:
    """Manage workflows — list, run, status."""

list_workflows 

list_workflows() -> None

List available workflow definitions.

Source code in src/diapason/cli/workflow_cmd.py 
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
@workflow.command("list")
def list_workflows() -> None:
    """List available workflow definitions."""
    console = Console(stderr=True)
    try:
        from diapason.workflow.loader import discover_workflows

        workflows = discover_workflows()
        if not workflows:
            console.print("[dim]No workflows found.[/dim]")
            return
        table = Table(title="Workflows")
        table.add_column("Name", style="cyan")
        table.add_column("Nodes", style="green")
        for name, wf in workflows.items():
            table.add_row(name, str(len(wf.nodes) if hasattr(wf, "nodes") else "?"))
        console.print(table)
    except ImportError:
        console.print("[dim]No workflows found. Define workflows in TOML files.[/dim]")
    except Exception as exc:
        console.print(f"[red]Error: {exc}[/red]")

run 

run(workflow_name: str, input_text: str | None) -> None

Run a workflow by name.

Source code in src/diapason/cli/workflow_cmd.py 
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
@workflow.command()
@click.argument("workflow_name")
@click.option("--input", "input_text", default=None, help="Input text for workflow.")
def run(workflow_name: str, input_text: str | None) -> None:
    """Run a workflow by name."""
    console = Console(stderr=True)
    console.print(f"[yellow]Running workflow: {workflow_name}[/yellow]")
    try:
        from diapason.workflow.loader import discover_workflows

        workflows = discover_workflows()
        if workflow_name not in workflows:
            console.print(f"[red]Workflow '{workflow_name}' not found.[/red]")
            return
        console.print(f"[green]Workflow '{workflow_name}' started.[/green]")
        # Full execution would need a DiapasonSystem — just report for now
        console.print(
            "[dim]Note: Full workflow execution requires a running system.[/dim]"
        )
    except ImportError:
        console.print("[red]Workflow system not available.[/red]")
    except Exception as exc:
        console.print(f"[red]Error: {exc}[/red]")

status 

status() -> None

Show status of running workflows.

Source code in src/diapason/cli/workflow_cmd.py 
63
64
65
66
67
@workflow.command()
def status() -> None:
    """Show status of running workflows."""
    console = Console(stderr=True)
    console.print("[dim]No workflows currently running.[/dim]")