update cli with new commands

Author: snopoke

taskbadger/__init__.py

@@ -1,7 +1,8 @@
 from .integrations import Action, EmailIntegration, WebhookIntegration
 from .internal.models import StatusEnum
+from .mug import Session
 from .safe_sdk import create_task_safe, update_task_safe
-from .sdk import Task, create_task, get_task, init, update_task
+from .sdk import DefaultMergeStrategy, Task, create_task, get_task, init, update_task
 
 try:
     import importlib.metadata as importlib_metadata

taskbadger/cli.py

@@ -0,0 +1,3 @@
+from .basics import create, get, update
+from .list_tasks import list_tasks_command
+from .wrapper import run

taskbadger/cli/__init__.py

@@ -0,0 +1,126 @@
+import csv
+import json
+import sys
+from typing import Tuple
+
+import typer
+from rich import print
+
+from taskbadger import StatusEnum, create_task, get_task, update_task
+from taskbadger.cli.utils import OutputFormat, configure_api, err_console, get_actions, get_metadata
+
+
+def get(
+    ctx: typer.Context,
+    task_id: str = typer.Argument(..., show_default=False, help="The ID of the task."),
+    output_format: OutputFormat = typer.Option(OutputFormat.pretty, "--format", "-f", help="Output format"),
+):
+    """Get a task."""
+    configure_api(ctx)
+    task = get_task(task_id)
+    if output_format == OutputFormat.pretty:
+        print(f"Task ID: {task.id}")
+        print(f"Created: {task.created.isoformat()}")
+        print(f"Name: {task.name}")
+        print(f"Status: {task.status}")
+        print(f"Percent: {task.value_percent}%")
+    elif output_format == OutputFormat.json:
+        print(json.dumps(task.to_dict(), indent=2))
+    elif output_format == OutputFormat.csv:
+        writer = csv.writer(sys.stdout)
+        writer.writerow("Task ID,Created,Name,Status,Percent".split(","))
+        writer.writerow([task.id, task.created.isoformat(), task.name, task.status, str(task.value_percent)])
+
+
+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:me@email.com'",
+    ),
+    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:me@email.com'",
+    ),
+    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/basics.py

@@ -0,0 +1,79 @@
+import csv
+import json
+import sys
+from urllib.parse import parse_qs, urlparse
+
+import typer
+from rich import print
+from rich.console import Console
+from rich.table import Table
+
+from taskbadger.cli.utils import OutputFormat, configure_api
+from taskbadger.sdk import list_tasks
+
+
+def list_tasks_command(
+    ctx: typer.Context,
+    output_format: OutputFormat = typer.Option(OutputFormat.pretty, "--format", "-f", help="Output format"),
+    limit: int = typer.Option(100, help="Limit the number of results."),
+    start_token: str = typer.Option(None, show_default=False, help="Start token."),
+):
+    """List tasks."""
+    configure_api(ctx)
+    tasks = list_tasks(page_size=limit, cursor=start_token)
+    render(output_format, ctx, tasks)
+
+
+def render(format_: OutputFormat, ctx, result):
+    if format_ == OutputFormat.pretty:
+        _render_pretty(ctx, result)
+    elif format_ == OutputFormat.json:
+        _render_json(ctx, result)
+    elif format_ == OutputFormat.csv:
+        _render_csv(ctx, result)
+    else:
+        raise ValueError(f"Unknown format: {format_}")
+
+
+def _render_pretty(ctx, result):
+    table = Table(
+        title=f"Project: {ctx.meta['tb_config'].project_slug}, Organization: {ctx.meta['tb_config'].organization_slug}"
+    )
+
+    table.add_column("Task ID", no_wrap=True)
+    table.add_column("Created", no_wrap=True)
+    table.add_column("Name")
+    table.add_column("Status", no_wrap=True)
+    table.add_column("Percent", no_wrap=True)
+
+    for task in result.results:
+        table.add_row(task.id, task.created.isoformat(), task.name, task.status, str(task.value_percent))
+    Console().print(table)
+
+    cursor = _get_cursor(result.next_)
+    if cursor:
+        print(f"Next page token: [bold green]'{cursor}'[/]")
+
+
+def _render_csv(ctx, result):
+    writer = csv.writer(sys.stdout)
+    writer.writerow("Task ID,Created,Name,Status,Percent".split(","))
+    for task in result.results:
+        writer.writerow([task.id, task.created.isoformat(), task.name, task.status, str(task.value_percent)])
+
+    cursor = _get_cursor(result.next_)
+    if cursor:
+        print(f"next_token,{cursor}")
+
+
+def _render_json(ctx, result):
+    obj = result.to_dict()
+    ret = {"next_token": _get_cursor(result.next_), "results": obj["results"]}
+    print(json.dumps(ret, indent=2))
+
+
+def _get_cursor(url):
+    if url:
+        qs = urlparse(url).query
+        query = parse_qs(qs)
+        return query["cursor"][0]