add create / update commands

pyproject.toml

@@ -55,7 +55,7 @@ invoke = "^2.0.0"
 pytest-celery = "^0.0.0"
 
 [tool.poetry.scripts]
-taskbadger = "taskbadger.cli:app"
+taskbadger = "taskbadger.cli_main:app"
 
 [build-system]
 requires = ["poetry-core>=1.0.0"]

taskbadger/cli/__init__.py

@@ -1 +1,2 @@
 from .run import run
+from .task import create, update

taskbadger/cli/run.py

@@ -3,14 +3,14 @@
 import typer
 from rich import print
 
-from taskbadger import Action, DefaultMergeStrategy, Session, StatusEnum, Task, integrations
-from taskbadger.cli.utils import _configure_api, err_console
+from taskbadger import DefaultMergeStrategy, Session, StatusEnum, Task
+from taskbadger.cli.utils import configure_api, err_console, get_actions
 from taskbadger.process import ProcessRunner
 
 
 def run(
     ctx: typer.Context,
-    name: str,
+    name: str = typer.Argument(..., show_default=False, help="The task name"),
     monitor_id: str = typer.Option(None, help="Associate this task with a monitor."),
     update_frequency: int = typer.Option(5, metavar="SECONDS", min=5, max=300, help="Seconds between updates."),
     action_def: Tuple[str, str, str] = typer.Option(
@@ -34,19 +34,16 @@ def run(
 
         [on black]taskbadger run 'my task' -- ./my-script.sh arg -v[/]
     """
-    _configure_api(ctx)
-    action = None
-    if any(action_def):
-        trigger, integration, config = action_def
-        action = Action(trigger, integrations.from_config(integration, config))
+    configure_api(ctx)
+    actions = get_actions(action_def)
     stale_timeout = update_frequency * 2
     with Session():
         try:
             task = Task.create(
                 name,
                 status=StatusEnum.PROCESSING,
                 stale_timeout=stale_timeout,
-                actions=[action] if action else None,
+                actions=actions,
                 monitor_id=monitor_id,
             )
         except Exception as e:

taskbadger/cli/task.py

@@ -0,0 +1,101 @@
+from typing import Tuple
+
+import typer
+from rich import print
+
+from taskbadger import StatusEnum, create_task, update_task
+from taskbadger.cli.utils import configure_api, err_console, get_actions, get_metadata
+
+
+def create(
+    ctx: typer.Context,
+    name: str = typer.Argument(..., show_default=False, help="The task name."),
+    monitor_id: str = typer.Option(None, help="Associate this task with a monitor."),
+    action_def: Tuple[str, str, str] = typer.Option(
+        (None, None, None),
+        "--action",
+        "-a",
+        metavar="<trigger integration config>",
+        show_default=False,
+        help="Action definition e.g. 'success,error email to:[email protected]'",
+    ),
+    status: StatusEnum = typer.Option(StatusEnum.PROCESSING, help="The initial status of the task."),
+    value_max: int = typer.Option(100, help="The maximum value for the task."),
+    metadata: list[str] = typer.Option(
+        None,
+        show_default=False,
+        help="Metadata 'key=value' pair to associate with the task. Can be specified multiple times.",
+    ),
+    metadata_json: str = typer.Option(
+        None, show_default=False, help="Metadata to associate with the task. Must be valid JSON."
+    ),
+    quiet: bool = typer.Option(False, "--quiet", "-q", help="Minimal output. Only the Task ID."),
+):
+    """Create a task."""
+    configure_api(ctx)
+    actions = get_actions(action_def)
+    metadata = get_metadata(metadata, metadata_json)
+
+    try:
+        task = create_task(
+            name,
+            status=status,
+            value_max=value_max,
+            data=metadata,
+            actions=actions,
+            monitor_id=monitor_id,
+        )
+    except Exception as e:
+        err_console.print(f"Error creating task: {e}")
+    else:
+        if quiet:
+            print(task.id)
+        else:
+            print(f"Task created: {task.public_url}")
+
+
+def update(
+    ctx: typer.Context,
+    task_id: str = typer.Argument(..., show_default=False, help="The ID of the task to update."),
+    name: str = typer.Option(None, show_default=False, help="Update the name of the task."),
+    action_def: Tuple[str, str, str] = typer.Option(
+        (None, None, None),
+        "--action",
+        "-a",
+        metavar="<trigger integration config>",
+        show_default=False,
+        help="Action definition e.g. 'success,error email to:[email protected]'",
+    ),
+    status: StatusEnum = typer.Option(StatusEnum.PROCESSING, help="The status of the task."),
+    value: int = typer.Option(None, show_default=False, help="The current task value (progress)."),
+    value_max: int = typer.Option(None, show_default=False, help="The maximum value for the task."),
+    metadata: list[str] = typer.Option(
+        None,
+        show_default=False,
+        help="Metadata 'key=value' pair to associate with the task. Can be specified multiple times.",
+    ),
+    metadata_json: str = typer.Option(
+        None, show_default=False, help="Metadata to associate with the task. Must be valid JSON."
+    ),
+    quiet: bool = typer.Option(False, "--quiet", "-q", help="No output."),
+):
+    """Update a task."""
+    configure_api(ctx)
+    actions = get_actions(action_def)
+    metadata = get_metadata(metadata, metadata_json)
+
+    try:
+        task = update_task(
+            task_id,
+            name=name,
+            status=status,
+            value=value,
+            value_max=value_max,
+            data=metadata,
+            actions=actions,
+        )
+    except Exception as e:
+        err_console.print(f"Error creating task: {e}")
+    else:
+        if not quiet:
+            print(f"Task updated: {task.public_url}")

taskbadger/cli/utils.py

@@ -1,11 +1,15 @@
+import json
+from typing import Any, Optional
+
 import typer
 from rich import print
 from rich.console import Console
 
+from taskbadger import Action, integrations
 from taskbadger.exceptions import ConfigurationError
 
 
-def _configure_api(ctx):
+def configure_api(ctx):
     config = ctx.meta["tb_config"]
     try:
         config.init_api()
@@ -15,3 +19,22 @@ def _configure_api(ctx):
 
 
 err_console = Console(stderr=True)
+
+
+def get_actions(action_def: tuple[str, str, str]) -> list[Action]:
+    if any(action_def):
+        trigger, integration, config = action_def
+        return [Action(trigger, integrations.from_config(integration, config))]
+    return []
+
+
+def get_metadata(metadata_kv: list[str], metadata_json: str) -> dict:
+    metadata = {}
+    for kv in metadata_kv:
+        k, v = kv.strip().split("=", 1)
+        metadata[k] = v
+
+    if metadata_json:
+        metadata.update(json.loads(metadata_json))
+
+    return metadata

taskbadger/cli_main.py

@@ -4,7 +4,7 @@
 from rich import print
 
 from taskbadger import __version__
-from taskbadger.cli import run
+from taskbadger.cli import create, run, update
 from taskbadger.config import get_config, write_config
 
 app = typer.Typer(
@@ -14,6 +14,8 @@
 
 
 app.command(context_settings={"allow_extra_args": True, "ignore_unknown_options": False})(run)
+app.command(context_settings={"ignore_unknown_options": False})(create)
+app.command(context_settings={"ignore_unknown_options": False})(update)
 
 
 def version_callback(value: bool):

tests/test_cli_create_update.py

@@ -0,0 +1,68 @@
+import os
+from http import HTTPStatus
+from unittest import mock
+
+import pytest
+from typer.testing import CliRunner
+
+from taskbadger.cli_main import app
+from taskbadger.internal.models import (
+    PatchedTaskRequest,
+    PatchedTaskRequestData,
+    StatusEnum,
+    TaskRequest,
+    TaskRequestData,
+)
+from taskbadger.internal.types import UNSET, Response
+from taskbadger.mug import Badger
+from taskbadger.sdk import Task
+from tests.utils import task_for_test
+
+runner = CliRunner()
+
+
+@pytest.fixture(autouse=True)
+def mock_env():
+    with mock.patch.dict(
+        os.environ,
+        {
+            "TASKBADGER_ORG": "org",
+            "TASKBADGER_PROJECT": "project",
+            "TASKBADGER_API_KEY": "token",
+        },
+        clear=True,
+    ):
+        yield
+
+
+def test_cli_create():
+    with (mock.patch("taskbadger.sdk.task_create.sync_detailed") as create,):
+        task = task_for_test()
+        create.return_value = Response(HTTPStatus.OK, b"", {}, task)
+
+        args = ["create", "my-task", "--metadata-json", '{"a": 1, "c": 1}', "--metadata", "b=2", "--metadata", "a=3"]
+        result = runner.invoke(app, args, catch_exceptions=False)
+        assert result.exit_code == 0, result.output
+
+        request = TaskRequest(
+            name="my-task",
+            status=StatusEnum.PROCESSING,
+            value_max=100,
+            data=TaskRequestData.from_dict({"b": "2", "a": 1, "c": 1}),
+        )
+        create.assert_called_with(client=mock.ANY, organization_slug="org", project_slug="project", json_body=request)
+
+
+def test_cli_update():
+    with mock.patch("taskbadger.sdk.task_partial_update.sync_detailed") as update:
+        task = task_for_test()
+        update.return_value = Response(HTTPStatus.OK, b"", {}, task)
+
+        result = runner.invoke(app, ["update", "task123", "--status=success", "--value", "100"], catch_exceptions=False)
+        assert result.exit_code == 0, result.output
+
+        body = PatchedTaskRequest(status=StatusEnum.SUCCESS, value=100)
+
+        update.assert_called_with(
+            client=mock.ANY, organization_slug="org", project_slug="project", id="task123", json_body=body
+        )