Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 3 additions & 0 deletions src/keboola_agent_cli/cli.py
Original file line number Diff line number Diff line change
Expand Up @@ -10,6 +10,7 @@
from .commands.project import project_app
from .config_store import ConfigStore
from .output import OutputFormatter
from .services.config_service import ConfigService
from .services.project_service import ProjectService

app = typer.Typer(
Expand Down Expand Up @@ -58,6 +59,7 @@ def main(
config_store = ConfigStore()

project_service = ProjectService(config_store=config_store)
config_service = ConfigService(config_store=config_store)

ctx.ensure_object(dict)
ctx.obj["formatter"] = formatter
Expand All @@ -66,3 +68,4 @@ def main(
ctx.obj["no_color"] = effective_no_color
ctx.obj["config_store"] = config_store
ctx.obj["project_service"] = project_service
ctx.obj["config_service"] = config_service
90 changes: 84 additions & 6 deletions src/keboola_agent_cli/commands/config.py
Original file line number Diff line number Diff line change
@@ -1,33 +1,86 @@
"""Configuration browsing commands - list and detail."""
"""Configuration browsing commands - list and detail.

Thin CLI layer: parses arguments, calls ConfigService, formats output.
No business logic belongs here.
"""

from typing import Optional

import typer

from ..output import OutputFormatter
from ..errors import ConfigError, KeboolaApiError
from ..output import OutputFormatter, format_config_detail, format_configs_table
from ..services.config_service import ConfigService

config_app = typer.Typer(help="Browse and inspect configurations")

VALID_COMPONENT_TYPES = ["extractor", "writer", "transformation", "application"]


def _get_formatter(ctx: typer.Context) -> OutputFormatter:
"""Retrieve the OutputFormatter from the Typer context."""
return ctx.obj["formatter"]


def _get_service(ctx: typer.Context) -> ConfigService:
"""Retrieve the ConfigService from the Typer context."""
return ctx.obj["config_service"]


@config_app.command("list")
def config_list(
ctx: typer.Context,
project: Optional[list[str]] = typer.Option(None, "--project", help="Project alias (can be repeated)"),
project: Optional[list[str]] = typer.Option(
None,
"--project",
help="Project alias to query (can be repeated for multiple projects)",
),
component_type: Optional[str] = typer.Option(
None,
"--component-type",
help="Filter by component type: extractor, writer, transformation, application",
),
component_id: Optional[str] = typer.Option(None, "--component-id", help="Filter by specific component ID"),
component_id: Optional[str] = typer.Option(
None,
"--component-id",
help="Filter by specific component ID (e.g. keboola.ex-db-snowflake)",
),
) -> None:
"""List configurations from connected projects."""
formatter = _get_formatter(ctx)
formatter.output("Not yet implemented", lambda c, d: c.print(d))
service = _get_service(ctx)

# Validate component_type if provided
if component_type and component_type not in VALID_COMPONENT_TYPES:
formatter.error(
message=f"Invalid component type '{component_type}'. "
f"Valid types: {', '.join(VALID_COMPONENT_TYPES)}",
error_code="INVALID_ARGUMENT",
)
raise typer.Exit(code=2)

try:
result = service.list_configs(
aliases=project,
component_type=component_type,
component_id=component_id,
)
except ConfigError as exc:
formatter.error(message=exc.message, error_code="CONFIG_ERROR")
raise typer.Exit(code=5)

# In JSON mode, include both configs and errors in the response
if formatter.json_mode:
formatter.output(result)
else:
# In human mode, show per-project errors as warnings and configs as table
format_configs_table(formatter.console, result)

# Show error warnings on stderr too
for err in result.get("errors", []):
formatter.warning(
f"Project '{err['project_alias']}': {err['message']}"
)


@config_app.command("detail")
Expand All @@ -39,4 +92,29 @@ def config_detail(
) -> None:
"""Show detailed information about a specific configuration."""
formatter = _get_formatter(ctx)
formatter.output("Not yet implemented", lambda c, d: c.print(d))
service = _get_service(ctx)

try:
result = service.get_config_detail(
alias=project,
component_id=component_id,
config_id=config_id,
)
formatter.output(result, format_config_detail)
except ConfigError as exc:
formatter.error(message=exc.message, error_code="CONFIG_ERROR")
raise typer.Exit(code=5)
except KeboolaApiError as exc:
if exc.error_code == "INVALID_TOKEN":
exit_code = 3
elif exc.error_code in ("TIMEOUT", "CONNECTION_ERROR", "RETRY_EXHAUSTED"):
exit_code = 4
else:
exit_code = 1
formatter.error(
message=exc.message,
error_code=exc.error_code,
project=project,
retryable=exc.retryable,
)
raise typer.Exit(code=exit_code)
114 changes: 114 additions & 0 deletions src/keboola_agent_cli/output.py
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,8 @@
from typing import Any, Callable

from rich.console import Console
from rich.panel import Panel
from rich.table import Table

from .models import ErrorResponse, SuccessResponse

Expand Down Expand Up @@ -86,3 +88,115 @@ def success(self, message: str) -> None:
sys.stdout.write(response.model_dump_json(indent=2) + "\n")
else:
self.console.print(f"[bold green]Success:[/bold green] {message}")

def warning(self, message: str) -> None:
"""Output a warning message to stderr (human mode only).

In JSON mode, warnings are not printed separately -- they are
embedded in the structured response via the errors list.

Args:
message: The warning message to display.
"""
if not self.json_mode:
self.err_console.print(f"[bold yellow]Warning:[/bold yellow] {message}")


def format_configs_table(console: Console, data: dict[str, Any]) -> None:
"""Render a Rich table of configurations grouped by project alias.

Args:
console: Rich Console instance.
data: Dict with "configs" (list of config dicts) and "errors" (list of error dicts).
"""
configs = data.get("configs", [])
errors = data.get("errors", [])

# Show per-project errors as warnings
for err in errors:
console.print(
f"[bold yellow]Warning:[/bold yellow] Project [bold]{err['project_alias']}[/bold]: "
f"{err['message']}"
)

if not configs:
if not errors:
console.print("No configurations found. Use [bold]kbagent project add[/bold] to connect a project first.")
else:
console.print("No configurations retrieved (all projects failed).")
return

# Group configs by project alias
projects_order: list[str] = []
grouped: dict[str, list[dict[str, Any]]] = {}
for cfg in configs:
alias = cfg["project_alias"]
if alias not in grouped:
projects_order.append(alias)
grouped[alias] = []
grouped[alias].append(cfg)

for alias in projects_order:
project_configs = grouped[alias]
table = Table(title=f"Configurations - {alias}")
table.add_column("Component", style="bold cyan")
table.add_column("Type", style="dim")
table.add_column("Config ID", justify="right")
table.add_column("Config Name")
table.add_column("Description", style="dim", max_width=40)

for cfg in project_configs:
table.add_row(
cfg["component_id"],
cfg["component_type"],
cfg["config_id"],
cfg["config_name"],
cfg.get("config_description", ""),
)

console.print(table)
console.print()


def format_config_detail(console: Console, data: dict[str, Any]) -> None:
"""Render detailed configuration information.

Args:
console: Rich Console instance.
data: Configuration detail dict from the API.
"""
alias = data.get("project_alias", "unknown")
name = data.get("name", "Unknown")
config_id = data.get("id", "")
description = data.get("description", "")
component_id = data.get("component_id", data.get("componentId", ""))

header = f"Configuration Detail - {alias}"

lines = [
f"[bold]Name:[/bold] {name}",
f"[bold]Config ID:[/bold] {config_id}",
f"[bold]Component:[/bold] {component_id}",
]
if description:
lines.append(f"[bold]Description:[/bold] {description}")

# Show configuration parameters if present
configuration = data.get("configuration", {})
if configuration:
import json as _json
config_str = _json.dumps(configuration, indent=2)
lines.append(f"\n[bold]Configuration:[/bold]\n{config_str}")

# Show rows if present
rows = data.get("rows", [])
if rows:
lines.append(f"\n[bold]Rows:[/bold] {len(rows)} row(s)")
for row in rows[:10]: # Show at most 10 rows
row_name = row.get("name", row.get("id", ""))
lines.append(f" - {row_name}")
if len(rows) > 10:
lines.append(f" ... and {len(rows) - 10} more")

panel = Panel("\n".join(lines), title=header, expand=False)
console.print(panel)
Loading