wip

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

Author: ybastide

.pre-commit-config.yaml

@@ -14,14 +14,14 @@ repos:
           - --maxkb=1024
   - repo: https://github.com/astral-sh/uv-pre-commit
     # uv version.
-    rev: 0.5.24
+    rev: 0.6.6
     hooks:
       # Keep uv.lock up to date.
       - id: uv-lock
 
   - repo: https://github.com/astral-sh/ruff-pre-commit
     # Ruff version.
-    rev: v0.9.3
+    rev: v0.11.0
     hooks:
       # Run the linter.
       - id: ruff

pyproject.toml

@@ -46,6 +46,7 @@ dependencies = [
     "click",
     "psutil",
     "pytz",
+    "typer>=0.15.2",
 ]
 
 [project.urls]
@@ -92,7 +93,7 @@ exclude = ["*~"]
 [dependency-groups]
 dev = [
     "boto3-stubs[s3,swf]",
-    "cffi==v1.17.0rc1; python_full_version=='3.13.0b4'", # via cryptography via moto, secretstorage
+    "cffi==v1.17.1; python_full_version=='3.13.0b4'", # via cryptography via moto, secretstorage
     "flaky",
     "hatch",
     "invoke",

simpleflow/cli/__init__.py

@@ -0,0 +1,142 @@
+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.command import get_progression_callback, get_workflow_type, with_format
+from simpleflow.swf import helpers
+from simpleflow.swf.mapper.models import WorkflowExecution
+from simpleflow.swf.utils import set_workflow_class_name
+from simpleflow.utils import import_from_module
+
+
+class Status(str, Enum):
+    open = "open"
+    closed = "closed"
+
+
+class CloseStatus(str, Enum):
+    completed = "completed"
+    failed = "failed"
+    canceled = "canceled"
+    terminated = "terminated"
+    continued_as_new = "continued_as_new"
+
+
+app = typer.Typer(no_args_is_help=True)
+
+TIMESTAMP_FORMATS = [
+    "%Y-%m-%d",
+    "%Y-%m-%dT%H:%M:%S%z",
+    "%Y-%m-%d %H:%M:%S%z",
+    "%Y-%m-%dT%H:%M:%S",
+    "%Y-%m-%d %H:%M:%S",
+]
+
+
+@app.command()
+def filter(
+    ctx: typer.Context,
+    domain: Annotated[str, typer.Argument(envvar="SWF_DOMAIN")],
+    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,
+    started_since: int = 1,
+    from_date: Annotated[datetime, typer.Option(formats=TIMESTAMP_FORMATS)] | None = None,
+    to_date: Annotated[datetime, typer.Option(formats=TIMESTAMP_FORMATS)] | None = None,
+):
+    """
+    Filter workflow executions.
+    """
+    status = status.upper()
+    kwargs: dict[str, Any] = {}
+    if status == WorkflowExecution.STATUS_OPEN:
+        if from_date:
+            kwargs["oldest_date"] = from_date
+            kwargs["latest_date"] = to_date
+        else:
+            kwargs["oldest_date"] = started_since
+    else:
+        if from_date:
+            kwargs["start_oldest_date"] = from_date
+            kwargs["start_latest_date"] = to_date
+        else:
+            kwargs["start_oldest_date"] = started_since
+
+    if close_status and status != WorkflowExecution.STATUS_CLOSED:
+        raise Exception("Closed status not supported for non-closed workflows.")
+    elif close_status:
+        kwargs["close_status"] = close_status.upper()
+
+    print(
+        with_format(ctx.parent)(helpers.filter_workflow_executions)(
+            domain,
+            status=status.upper(),
+            tag=tag,
+            workflow_id=workflow_id,
+            workflow_type_name=workflow_type,
+            workflow_type_version=workflow_type_version,
+            callback=get_progression_callback("executionInfos"),
+            **kwargs,
+        )
+    )
+
+
+@app.command()
+def start(
+    ctx: typer.Context,
+    workflow: str,
+    domain: Annotated[str, typer.Argument(envvar="SWF_DOMAIN")],
+    input: Annotated[str, typer.Option("--input", "-i", help="input JSON")] | None = None,
+):
+    """
+    Start a workflow.
+    """
+    workflow_class = import_from_module(workflow)
+    wf_input: dict[str, Any] = {}
+    if input is not None:
+        json_input = json.loads(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):
+            wf_input = {"args": [], "kwargs": json_input}
+        else:
+            wf_input = json_input
+    workflow_type = get_workflow_type(domain, workflow_class)
+    set_workflow_class_name(wf_input, workflow_class)
+    execution = workflow_type.start_execution(
+        # workflow_id=workflow_id,
+        # task_list=task_list or workflow_class.task_list,
+        # execution_timeout=execution_timeout,
+        input=wf_input,
+        # tag_list=tags,
+        # decision_tasks_timeout=decision_tasks_timeout,
+    )
+
+    def get_infos():
+        return ["workflow_id", "run_id"], [[execution.workflow_id, execution.run_id]]
+
+    print(with_format(ctx.parent)(get_infos)())
+
+
+if __name__ == "__main__":
+    # from click.core import Command
+    #
+    # parent = typer.Context(command=Command(name="main"))
+    # parent.params["format"] = "json"
+    # filter(
+    #     ctx=typer.Context(
+    #         command=Command(name="filter"), parent=typer.Context(command=Command(name="main"), parent=parent)
+    #     ),
+    #     domain="TestDomain",
+    # )
+    app()

simpleflow/cli/workflow.py

@@ -271,9 +271,14 @@ def restart_workflow(domain: str, workflow_id: str, run_id: str | None):
 
 
 def with_format(ctx):
+    with_header = ctx.parent.params.get("header")
+    fmt = ctx.parent.params.get("format") or pretty.DEFAULT_FORMAT
+    if fmt == "prettyjson":
+        with_header = True
+        fmt = "json"
     return pretty.formatted(
-        with_header=ctx.parent.params["header"],
-        fmt=ctx.parent.params["format"] or pretty.DEFAULT_FORMAT,
+        with_header=with_header,
+        fmt=fmt,
     )
 
 

simpleflow/command.py

@@ -0,0 +1,52 @@
+#!/usr/bin/env python3
+from __future__ import annotations
+
+from enum import Enum
+
+import typer
+from typing_extensions import Annotated
+
+from simpleflow.cli import workflow
+
+app = typer.Typer(
+    # add_completion=False,
+    no_args_is_help=True,
+    context_settings={"help_option_names": ["--help", "-h"]},
+)
+
+app.add_typer(workflow.app, name="workflow", help="Manage workflows")
+
+
+class Format(str, Enum):
+    json = "json"
+    prettyjson = "prettyjson"
+    csv = "csv"
+    tsv = "tsv"
+    tabular = "tabular"
+    human = "human"
+
+
+@app.callback()
+def main(
+    ctx: typer.Context,
+    format: Annotated[Format, typer.Option("--format", "-f", envvar="SIMPLEFLOW_FORMAT")] = Format.json,
+):
+    ctx.params["format"] = format.lower()
+
+
+# @app.command()
+# def main(name: str):
+#     print(f"Hello {name}")
+#
+#
+# def run(function: Callable[..., Any]) -> None:
+#     app = Typer(
+#         # add_completion=False
+#     )
+#     app.command(context_settings={"help_option_names": ["-h", "--help"]})(function)
+#     app()
+
+
+if __name__ == "__main__":
+    app()
+    # run(main)

simpleflow/main.py

@@ -251,7 +251,7 @@ def start_execution(
         :param  decision_tasks_timeout: maximum duration of decision tasks
                                         for this workflow execution
         """
-        workflow_id = workflow_id or f"{self.name}-{self.version}-{time.time():d}"
+        workflow_id = workflow_id or f"{self.name}-{self.version}-{int(time.time())}"
         task_list = task_list or self.task_list
         child_policy = child_policy or self.child_policy
         if child_policy not in CHILD_POLICIES: