feat(infrahubctl): add task list command

docs/docs/infrahubctl/infrahubctl-task.mdx

@@ -0,0 +1,41 @@
+# `infrahubctl task`
+
+Manage Infrahub tasks.
+
+**Usage**:
+
+```console
+$ infrahubctl task [OPTIONS] COMMAND [ARGS]...
+```
+
+**Options**:
+
+* `--install-completion`: Install completion for the current shell.
+* `--show-completion`: Show completion for the current shell, to copy it or customize the installation.
+* `--help`: Show this message and exit.
+
+**Commands**:
+
+* `list`: List Infrahub tasks.
+
+## `infrahubctl task list`
+
+List Infrahub tasks.
+
+**Usage**:
+
+```console
+$ infrahubctl task list [OPTIONS]
+```
+
+**Options**:
+
+* `-s, --state TEXT`: Filter by task state. Can be provided multiple times.
+* `--limit INTEGER`: Maximum number of tasks to retrieve.
+* `--offset INTEGER`: Offset for pagination.
+* `--include-related-nodes / --no-include-related-nodes`: Include related nodes in the output.  [default: no-include-related-nodes]
+* `--include-logs / --no-include-logs`: Include task logs in the output.  [default: no-include-logs]
+* `--json`: Output the result as JSON.
+* `--debug / --no-debug`: [default: no-debug]
+* `--config-file TEXT`: [env var: INFRAHUBCTL_CONFIG; default: infrahubctl.toml]
+* `--help`: Show this message and exit.

docs/sidebars-infrahubctl.ts

@@ -24,6 +24,7 @@ const sidebars: SidebarsConfig = {
         'infrahubctl-repository',
         'infrahubctl-run',
         'infrahubctl-schema',
+        'infrahubctl-task',
         'infrahubctl-transform',
         'infrahubctl-validate',
         'infrahubctl-version'

infrahub_sdk/ctl/cli_commands.py

@@ -32,6 +32,7 @@
 from ..ctl.repository import app as repository_app
 from ..ctl.repository import get_repository_config
 from ..ctl.schema import app as schema_app
+from ..ctl.task import app as task_app
 from ..ctl.transform import list_transforms
 from ..ctl.utils import (
     catch_exception,
@@ -63,6 +64,7 @@
 app.add_typer(repository_app, name="repository")
 app.add_typer(menu_app, name="menu")
 app.add_typer(object_app, name="object")
+app.add_typer(task_app, name="task")
 
 app.command(name="dump")(dump)
 app.command(name="load")(load)

infrahub_sdk/ctl/task.py

@@ -0,0 +1,110 @@
+from __future__ import annotations
+
+from typing import Optional
+
+import typer
+from rich.console import Console
+from rich.table import Table
+
+from ..async_typer import AsyncTyper
+from ..task.manager import TaskFilter
+from ..task.models import Task, TaskState
+from .client import initialize_client
+from .parameters import CONFIG_PARAM
+from .utils import catch_exception, init_logging
+
+app = AsyncTyper()
+console = Console()
+
+
+@app.callback()
+def callback() -> None:
+    """Manage Infrahub tasks."""
+
+
+def _parse_states(states: list[str] | None) -> list[TaskState] | None:
+    if not states:
+        return None
+
+    parsed_states: list[TaskState] = []
+    for state in states:
+        normalized_state = state.strip().upper()
+        try:
+            parsed_states.append(TaskState(normalized_state))
+        except ValueError as exc:  # pragma: no cover - typer will surface this as CLI error
+            raise typer.BadParameter(
+                f"Unsupported state '{state}'. Available states: {', '.join(item.value.lower() for item in TaskState)}"
+            ) from exc
+
+    return parsed_states
+
+
+def _render_table(tasks: list[Task]) -> None:
+    table = Table(title="Infrahub Tasks", box=None)
+    table.add_column("ID", style="cyan", overflow="fold")
+    table.add_column("Title", style="magenta", overflow="fold")
+    table.add_column("State", style="green")
+    table.add_column("Progress", justify="right")
+    table.add_column("Workflow", overflow="fold")
+    table.add_column("Branch", overflow="fold")
+    table.add_column("Updated")
+
+    if not tasks:
+        table.add_row("-", "No tasks found", "-", "-", "-", "-", "-")
+        console.print(table)
+        return
+
+    for task in tasks:
+        progress = f"{task.progress:.0%}" if task.progress is not None else "-"
+        table.add_row(
+            task.id,
+            task.title,
+            task.state.value,
+            progress,
+            task.workflow or "-",
+            task.branch or "-",
+            task.updated_at.isoformat(),
+        )
+
+    console.print(table)
+
+
+@app.command(name="list")
+@catch_exception(console=console)
+async def list_tasks(
+    state: list[str] = typer.Option(
+        None, "--state", "-s", help="Filter by task state. Can be provided multiple times."
+    ),
+    limit: Optional[int] = typer.Option(None, help="Maximum number of tasks to retrieve."),
+    offset: Optional[int] = typer.Option(None, help="Offset for pagination."),
+    include_related_nodes: bool = typer.Option(False, help="Include related nodes in the output."),
+    include_logs: bool = typer.Option(False, help="Include task logs in the output."),
+    json_output: bool = typer.Option(False, "--json", help="Output the result as JSON."),
+    debug: bool = False,
+    _: str = CONFIG_PARAM,
+) -> None:
+    """List Infrahub tasks."""
+
+    init_logging(debug=debug)
+
+    client = initialize_client()
+    filters = TaskFilter()
+    parsed_states = _parse_states(state)
+    if parsed_states:
+        filters.state = parsed_states
+
+    tasks = await client.task.filter(
+        filter=filters,
+        limit=limit,
+        offset=offset,
+        include_related_nodes=include_related_nodes,
+        include_logs=include_logs,
+    )
+
+    if json_output:
+        console.print_json(
+            data=[task.model_dump(mode="json") for task in tasks], indent=2, sort_keys=True, highlight=False
+        )
+        return
+
+    _render_table(tasks)

infrahub_sdk/task/models.py

@@ -49,12 +49,14 @@ def from_graphql(cls, data: dict) -> Task:
         related_nodes: list[TaskRelatedNode] = []
         logs: list[TaskLog] = []
 
-        if data.get("related_nodes"):
-            related_nodes = [TaskRelatedNode(**item) for item in data["related_nodes"]]
+        if "related_nodes" in data:
+            if data.get("related_nodes"):
+                related_nodes = [TaskRelatedNode(**item) for item in data["related_nodes"]]
             del data["related_nodes"]
 
-        if data.get("logs"):
-            logs = [TaskLog(**item["node"]) for item in data["logs"]["edges"]]
+        if "logs" in data:
+            if data.get("logs"):
+                logs = [TaskLog(**item["node"]) for item in data["logs"]["edges"]]
             del data["logs"]
 
         return cls(**data, related_nodes=related_nodes, logs=logs)

tests/unit/ctl/test_task_app.py

@@ -0,0 +1,100 @@
+from __future__ import annotations
+
+import json
+from datetime import datetime, timezone
+from typing import TYPE_CHECKING
+
+import pytest
+from typer.testing import CliRunner
+
+from infrahub_sdk.ctl.task import app
+
+runner = CliRunner()
+
+pytestmark = pytest.mark.httpx_mock(can_send_already_matched_responses=True)
+
+if TYPE_CHECKING:
+    from pytest_httpx import HTTPXMock
+
+
+def _task_response() -> dict:
+    now = datetime(2024, 1, 1, 12, 0, 0, tzinfo=timezone.utc).isoformat()
+    return {
+        "data": {
+            "InfrahubTask": {
+                "edges": [
+                    {
+                        "node": {
+                            "id": "task-1",
+                            "title": "Sync repositories",
+                            "state": "RUNNING",
+                            "progress": 0.5,
+                            "workflow": "RepositorySync",
+                            "branch": "main",
+                            "created_at": now,
+                            "updated_at": now,
+                            "logs": {"edges": []},
+                            "related_nodes": [],
+                        }
+                    }
+                ],
+                "count": 1,
+            }
+        }
+    }
+
+
+def _empty_task_response() -> dict:
+    return {"data": {"InfrahubTask": {"edges": [], "count": 0}}}
+
+
+def test_task_list_command(httpx_mock: HTTPXMock) -> None:
+    httpx_mock.add_response(
+        method="POST",
+        url="http://mock/graphql/main",
+        json=_task_response(),
+        match_headers={"X-Infrahub-Tracker": "query-tasks-page1"},
+    )
+
+    result = runner.invoke(app=app, args=["list"])
+
+    assert result.exit_code == 0
+    assert "Infrahub Tasks" in result.stdout
+    assert "Sync repositories" in result.stdout
+
+
+def test_task_list_json_output(httpx_mock: HTTPXMock) -> None:
+    httpx_mock.add_response(
+        method="POST",
+        url="http://mock/graphql/main",
+        json=_task_response(),
+        match_headers={"X-Infrahub-Tracker": "query-tasks-page1"},
+    )
+
+    result = runner.invoke(app=app, args=["list", "--json"])
+
+    assert result.exit_code == 0
+    payload = json.loads(result.stdout)
+    assert payload[0]["state"] == "RUNNING"
+    assert payload[0]["title"] == "Sync repositories"
+
+
+def test_task_list_with_state_filter(httpx_mock: HTTPXMock) -> None:
+    httpx_mock.add_response(
+        method="POST",
+        url="http://mock/graphql/main",
+        json=_empty_task_response(),
+        match_headers={"X-Infrahub-Tracker": "query-tasks-page1"},
+    )
+
+    result = runner.invoke(app=app, args=["list", "--state", "running"])
+
+    assert result.exit_code == 0
+    assert "No tasks found" in result.stdout
+
+
+def test_task_list_invalid_state() -> None:
+    result = runner.invoke(app=app, args=["list", "--state", "invalid"])
+
+    assert result.exit_code != 0
+    assert "Unsupported state" in result.stdout