wip

Signed-off-by: Yves Bastide <yves@botify.com>

Author: ybastide

simpleflow/cli/decider.py

@@ -0,0 +1,34 @@
+from __future__ import annotations
+
+import typer
+from typing_extensions import Annotated
+
+from simpleflow.swf.process.decider import command
+
+app = typer.Typer(no_args_is_help=True)
+
+
+@app.command()
+def start(
+    ctx: typer.Context,
+    workflows: Annotated[list[str] | None, typer.Argument()] = None,
+    *,
+    domain: Annotated[str, typer.Option(envvar="SWF_DOMAIN")],
+    task_list: Annotated[str, typer.Option("--task-list", "-t")] | None = None,
+    nb_processes: Annotated[int, typer.Option("--nb-processes", "-n")] | None = None,
+):
+    """
+    Start a decider.
+    """
+    if not workflows and not task_list:
+        raise typer.BadParameter("workflows or task_list is required")
+    command.start(
+        workflows=workflows or [],
+        domain=domain,
+        task_list=task_list,
+        nb_processes=nb_processes,
+    )
+
+
+if __name__ == "__main__":
+    app()

simpleflow/cli/workflow.py

@@ -1,13 +1,13 @@
 from __future__ import annotations
 
-import json
 from datetime import datetime
 from enum import Enum
 from typing import Any
 
 import typer
 from typing_extensions import Annotated
 
+from simpleflow import Workflow, format
 from simpleflow.command import get_progression_callback, get_workflow_type, with_format
 from simpleflow.swf import helpers
 from simpleflow.swf.mapper.models import WorkflowExecution
@@ -46,7 +46,6 @@ def filter(
     status: Annotated[Status, typer.Option("--status", "-s")] = Status.open,
     tag: str | None = None,
     workflow_id: str | None = None,
-    run_id: str | None = None,
     workflow_type: str | None = None,
     workflow_type_version: str | None = None,
     close_status: CloseStatus | None = None,
@@ -80,7 +79,7 @@ def filter(
     print(
         with_format(ctx.parent)(helpers.filter_workflow_executions)(
             domain,
-            status=status.upper(),
+            status=status,
             tag=tag,
             workflow_id=workflow_id,
             workflow_type_name=workflow_type,
@@ -95,16 +94,16 @@ def filter(
 def start(
     ctx: typer.Context,
     workflow: str,
-    domain: Annotated[str, typer.Argument(envvar="SWF_DOMAIN")],
+    domain: Annotated[str, typer.Option(envvar="SWF_DOMAIN")],
     input: Annotated[str, typer.Option("--input", "-i", help="input JSON")] | None = None,
 ):
     """
     Start a workflow.
     """
-    workflow_class = import_from_module(workflow)
+    workflow_class: type[Workflow] = import_from_module(workflow)
     wf_input: dict[str, Any] = {}
     if input is not None:
-        json_input = json.loads(input)
+        json_input = format.decode(input)
         if isinstance(json_input, list):
             wf_input = {"args": json_input, "kwargs": {}}
         elif isinstance(json_input, dict) and ("args" not in json_input or "kwargs" not in json_input):
@@ -113,9 +112,22 @@ def start(
             wf_input = json_input
     workflow_type = get_workflow_type(domain, workflow_class)
     set_workflow_class_name(wf_input, workflow_class)
+    get_task_list = getattr(workflow_class, "get_task_list", None)
+    if get_task_list:
+        if not callable(get_task_list):
+            raise Exception("get_task_list must be a callable")
+        if isinstance(wf_input, dict):
+            args = wf_input.get("args", [])
+            kwargs = wf_input.get("kwargs", {})
+        else:
+            args = []
+            kwargs = wf_input
+        task_list = get_task_list(workflow_class, *args, **kwargs)
+    else:
+        task_list = workflow_class.task_list
     execution = workflow_type.start_execution(
         # workflow_id=workflow_id,
-        # task_list=task_list or workflow_class.task_list,
+        task_list=task_list,
         # execution_timeout=execution_timeout,
         input=wf_input,
         # tag_list=tags,

simpleflow/main.py

@@ -6,7 +6,7 @@
 import typer
 from typing_extensions import Annotated
 
-from simpleflow.cli import workflow
+from simpleflow.cli import decider, workflow
 
 app = typer.Typer(
     # add_completion=False,
@@ -15,6 +15,7 @@
 )
 
 app.add_typer(workflow.app, name="workflow", help="Manage workflows")
+app.add_typer(decider.app, name="decider", help="Manage deciders")
 
 
 class Format(str, Enum):

simpleflow/swf/stats/pretty.py

@@ -232,6 +232,7 @@ def list_executions(
 def list_details(
     workflow_executions: list[WorkflowExecution],
 ) -> tuple[Sequence, Sequence]:
+    # FIXME: input is missing, what else?
     header = (
         "Workflow ID",
         "Workflow Type",
@@ -244,7 +245,7 @@ def list_details(
         "Close Timestamp",
         "Cancel Requested",
         "Execution Timeout",
-        "Input",
+        # "Input",
         "Tags",
         "Decision Tasks Timeout",
         "Parent Workflow ID",
@@ -264,7 +265,7 @@ def list_details(
             execution.close_timestamp,
             execution.cancel_requested,
             execution.execution_timeout,
-            execution.input,
+            # execution.input,
             execution.tag_list,
             execution.decision_tasks_timeout,
             execution.parent.get("workflowId"),