diff --git a/CLAUDE.md b/CLAUDE.md index 9d3103b0..ef4356b0 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -329,6 +329,15 @@ kbagent kai ask --message "question" [--project NAME] kbagent kai chat --message "msg" [--chat-id ID] [--project NAME] kbagent kai history [--project NAME] [--limit N] +kbagent flow list [--project NAME] [--branch ID] +kbagent flow detail --project NAME --flow-id ID [--component-id keboola.orchestrator|keboola.flow] [--branch ID] +kbagent flow schema +kbagent flow new --project NAME --name NAME [--component-id keboola.orchestrator|keboola.flow] [--description D] [--file @path.yaml|-|JSON] [--branch ID] +kbagent flow update --project NAME --flow-id ID [--component-id ID] [--name N] [--description D] [--file @path.yaml|-|JSON] [--branch ID] +kbagent flow delete --project NAME --flow-id ID [--component-id ID] [--branch ID] [--yes] +kbagent flow schedule --project NAME --flow-id ID --cron "0 6 * * *" [--component-id ID] [--timezone TZ] [--disabled] [--branch ID] +kbagent flow schedule-remove --project NAME --flow-id ID [--component-id ID] [--branch ID] [--yes] + kbagent context kbagent init [--from-global] kbagent doctor [--fix] diff --git a/plugins/kbagent/.claude-plugin/plugin.json b/plugins/kbagent/.claude-plugin/plugin.json index 8ec26950..d4cd469c 100644 --- a/plugins/kbagent/.claude-plugin/plugin.json +++ b/plugins/kbagent/.claude-plugin/plugin.json @@ -1,6 +1,6 @@ { "name": "kbagent", - "version": "0.21.1", + "version": "0.22.0", "description": "AI-friendly interface to Keboola Connection projects — explore configs, jobs, lineage, call MCP tools, manage dev branches, and debug SQL in workspaces", "author": { "name": "Keboola", diff --git a/plugins/kbagent/skills/kbagent/SKILL.md b/plugins/kbagent/skills/kbagent/SKILL.md index 3050395d..e839ae6f 100644 --- a/plugins/kbagent/skills/kbagent/SKILL.md +++ b/plugins/kbagent/skills/kbagent/SKILL.md @@ -133,6 +133,14 @@ When working inside a git repository or project directory, run `kbagent init` (o | Ask Kai a one-shot question and get the full response | `kbagent kai ask --message MESSAGE` | | Send a message to Kai in a chat session | `kbagent kai chat --message MESSAGE` | | List recent Kai chat sessions | `kbagent kai history` | +| List all flows (keboola.orchestrator + keboola.flow) across projects | `kbagent flow list` | +| Show detailed flow information including phases and tasks | `kbagent flow detail --project PROJECT --flow-id FLOW-ID` | +| Print the YAML format expected by 'flow new' and 'flow update' | `kbagent flow schema` | +| Create a new flow configuration | `kbagent flow new --project PROJECT --name NAME` | +| Update a flow's name, description, or phases/tasks | `kbagent flow update --project PROJECT --flow-id FLOW-ID` | +| Delete a flow configuration | `kbagent flow delete --project PROJECT --flow-id FLOW-ID` | +| Bind a cron schedule to a flow (upsert: creates or updates) | `kbagent flow schedule --project PROJECT --flow-id FLOW-ID --cron CRON` | +| Remove all schedules bound to a flow (deletes keboola.scheduler configs) | `kbagent flow schedule-remove --project PROJECT --flow-id FLOW-ID` | | List development branches from connected projects | `kbagent branch list` | | Create a new development branch and auto-activate it | `kbagent branch create --project PROJECT --name NAME` | | Set an existing development branch as active | `kbagent branch use --project PROJECT --branch BRANCH` | diff --git a/plugins/kbagent/skills/kbagent/references/commands-reference.md b/plugins/kbagent/skills/kbagent/references/commands-reference.md index 06ed9679..3e917723 100644 --- a/plugins/kbagent/skills/kbagent/references/commands-reference.md +++ b/plugins/kbagent/skills/kbagent/references/commands-reference.md @@ -108,6 +108,16 @@ All commands support `--json` for structured output. Multi-project flags (`--pro - `kai chat --message "msg" [--chat-id ID] [--project NAME]` -- send message in a chat session, returns chat_id for continuation - `kai history [--project NAME] [--limit N]` -- list recent Kai chat sessions (default limit: 10) +## Flows (Orchestrator) +- `flow list [--project NAME] [--branch ID]` -- list all flows (keboola.orchestrator + keboola.flow) across one or all projects +- `flow detail --project NAME --flow-id ID [--component-id keboola.orchestrator|keboola.flow] [--branch ID]` -- full phase/task breakdown; groups tasks by phase, lists orphan tasks +- `flow schema` -- print YAML template for flow configuration (phases + tasks); use with `--file @-` or save to a file +- `flow new --project NAME --name NAME [--component-id keboola.orchestrator|keboola.flow] [--description D] [--file @path.yaml|-|JSON] [--branch ID]` -- create a flow; DAG validated before API call; default component: keboola.flow +- `flow update --project NAME --flow-id ID [--component-id ID] [--name N] [--description D] [--file @path.yaml|-|JSON] [--branch ID]` -- update name, description, or phases/tasks; requires at least one of --name/--description/--file +- `flow delete --project NAME --flow-id ID [--component-id ID] [--branch ID] [--yes]` -- delete a flow config (confirmation guard) +- `flow schedule --project NAME --flow-id ID --cron "0 6 * * *" [--component-id ID] [--timezone TZ] [--disabled] [--branch ID]` -- attach a cron schedule (stored as keboola.scheduler config); replaces any existing schedule +- `flow schedule-remove --project NAME --flow-id ID [--component-id ID] [--branch ID] [--yes]` -- remove all cron schedules attached to a flow; idempotent + ## Sync (GitOps) - `sync init --project ALIAS [--directory DIR] [--git-branching]` -- initialize sync working directory - `sync pull --project ALIAS [--all-projects] [--force] [--dry-run] [--with-samples] [--no-storage] [--no-jobs] [--job-limit N]` -- download configs to local files. For large projects (>100 configs), automatically fetches jobs per-config when the grouped API limit is insufficient diff --git a/plugins/kbagent/skills/kbagent/references/flow-workflow.md b/plugins/kbagent/skills/kbagent/references/flow-workflow.md new file mode 100644 index 00000000..4b2e2715 --- /dev/null +++ b/plugins/kbagent/skills/kbagent/references/flow-workflow.md @@ -0,0 +1,110 @@ +# Flow Workflow + +Flows orchestrate Keboola transformations and extractors in a directed acyclic graph (DAG) of phases and tasks. kbagent supports two flow component types: `keboola.orchestrator` (classic) and `keboola.flow` (new format). + +## Core concepts + +- **Phase**: a named stage with `id` and `dependsOn` (list of upstream phase IDs). Phases with no `dependsOn` run first. +- **Task**: a unit of work referencing a component config, assigned to a phase via `phase` field. +- **Schedule**: stored as a `keboola.scheduler` config that targets the flow; not part of the flow config itself. + +## Quick start: create a flow + +```bash +# 1. See the template +kbagent flow schema + +# 2. Create a simple flow from YAML +cat > flow.yaml <<'EOF' +phases: + - id: 1 + name: Extract + dependsOn: [] + - id: 2 + name: Transform + dependsOn: [1] +tasks: + - id: 1 + name: Run extractor + phase: 1 + task: + mode: run + componentId: keboola.ex-db-snowflake + configId: "123456" + - id: 2 + name: Run transformation + phase: 2 + task: + mode: run + componentId: keboola.snowflake-transformation + configId: "789012" +EOF + +kbagent --json flow new --project prod --name "Daily ETL" --file @flow.yaml +``` + +## List and inspect flows + +```bash +# All flows across all projects +kbagent --json flow list + +# Flows in one project +kbagent --json flow list --project prod + +# Full phase/task breakdown +kbagent --json flow detail --project prod --flow-id 111 +``` + +## Update a flow + +```bash +# Rename only +kbagent --json flow update --project prod --flow-id 111 --name "New Name" + +# Replace phases/tasks from file (validates DAG before write) +kbagent --json flow update --project prod --flow-id 111 --file @updated.yaml +``` + +## Schedule a flow + +Schedules are stored as `keboola.scheduler` configs pointing at the flow. `flow schedule` is an upsert — if a schedule already exists for the flow it is updated in-place; otherwise a new one is created. Calling it twice with different cron expressions replaces the existing schedule. + +```bash +# Daily at 06:00 UTC +kbagent --json flow schedule --project prod --flow-id 111 --cron "0 6 * * *" + +# With timezone and disabled state +kbagent --json flow schedule \ + --project prod --flow-id 111 \ + --cron "0 8 * * 1-5" \ + --timezone "Europe/Prague" \ + --disabled + +# Remove all schedules (idempotent) +kbagent --json flow schedule-remove --project prod --flow-id 111 --yes +``` + +## Delete a flow + +```bash +kbagent --json flow delete --project prod --flow-id 111 --yes +``` + +## DAG validation + +kbagent validates the phase graph client-side before every create/update: +- Unknown `dependsOn` phase IDs → `INVALID_FLOW_DAG` +- Tasks referencing unknown phase IDs → `INVALID_FLOW_DAG` +- Cycles in the phase graph → `INVALID_FLOW_DAG` + +The error carries a list of human-readable violation messages. + +## Component IDs + +| Component | Use case | +|---|---| +| `keboola.flow` | New projects, preferred for new flows (default for `flow new`) | +| `keboola.orchestrator` | Legacy flows; most existing orchestrations use this (default for `flow detail/update/delete/schedule`) | + +Both are fully supported. Use `--component-id` to override the default. diff --git a/plugins/kbagent/skills/kbagent/references/gotchas.md b/plugins/kbagent/skills/kbagent/references/gotchas.md index c0d7b08d..1bf19e69 100644 --- a/plugins/kbagent/skills/kbagent/references/gotchas.md +++ b/plugins/kbagent/skills/kbagent/references/gotchas.md @@ -506,3 +506,30 @@ CLI hides via its four-bucket response, but they matter when interpreting result addressing. When exporting multiple tables, each ends up in a predictable subdirectory and there is no risk of name collisions. Override with `--output DIR` if you need a custom location. + +## Flow: default `--component-id` differs between commands + +- `kbagent flow new` defaults to **`keboola.flow`** (the newer format). +- `kbagent flow detail / update / delete / schedule / schedule-remove` all + default to **`keboola.orchestrator`** (the legacy format, since most + existing flows still use it). +- Consequence: if you create a flow with `flow new` and then call + `flow detail` without `--component-id`, you will get a `NOT_FOUND` error + because kbagent looks up the ID under `keboola.orchestrator`. Always pass + `--component-id keboola.flow` when round-tripping a flow you just created + via `flow new` (or, equivalently, pass `--component-id keboola.orchestrator` + on `flow new` to keep things consistent). +- `flow list` returns both component IDs and surfaces `component_id` on each + row — use it to confirm which variant a flow lives under before issuing + detail/update/delete/schedule commands. + +## Flow: `schedule` is an upsert (no `schedule-update`) + +- `kbagent flow schedule` creates a `keboola.scheduler` config on first run + and **updates the existing one in-place** on subsequent runs. Running it + twice with different `--cron` values replaces the schedule — it does not + create a second one. That's why there is no separate `flow schedule-update` + command. +- To inspect or remove schedules: `kbagent flow schedule-remove` deletes all + scheduler configs that target the flow. Pair it with `--dry-run` to see the + affected configs (cron + timezone) without calling `delete_config`. diff --git a/pyproject.toml b/pyproject.toml index 0eee9ad7..d622abf9 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "keboola-agent-cli" -version = "0.21.1" +version = "0.22.0" description = "AI-friendly CLI for managing Keboola projects" readme = "README.md" requires-python = ">=3.12" diff --git a/src/keboola_agent_cli/changelog.py b/src/keboola_agent_cli/changelog.py index 29f65c04..469416ae 100644 --- a/src/keboola_agent_cli/changelog.py +++ b/src/keboola_agent_cli/changelog.py @@ -8,6 +8,16 @@ # Ordered newest-first. Each value is a list of brief one-line descriptions. CHANGELOG: dict[str, list[str]] = { + "0.22.0": [ + "New: `kbagent flow list` -- list all flows (keboola.orchestrator + keboola.flow) across one or all projects; supports --project, --branch", + "New: `kbagent flow detail` -- full phase/task breakdown for a single flow config, including phase dependency graph and orphan detection", + "New: `kbagent flow schema` -- print the YAML template for flow configuration (phases + tasks) for use with --file", + "New: `kbagent flow new` -- create a flow with optional phases/tasks from a YAML/JSON --file; validates DAG before create", + "New: `kbagent flow update` -- update flow name, description, or phases/tasks; validates DAG before write; fetches current config before partial update", + "New: `kbagent flow delete` -- delete a flow config with --yes confirmation guard", + "New: `kbagent flow schedule` -- attach a cron schedule via keboola.scheduler; supports timezone and enabled/disabled state", + "New: `kbagent flow schedule-remove` -- remove all cron schedules attached to a flow; idempotent, --yes confirmation guard", + ], "0.21.1": [ "Fix: sync pull on a newly created dev branch now writes config rows (#193) -- idempotent skip guard for rows was missing a file-existence check, causing rows to be silently skipped when the branch directory was new (hash matched main because the branch is a clone)", ], diff --git a/src/keboola_agent_cli/cli.py b/src/keboola_agent_cli/cli.py index aeff431e..7ab545ee 100644 --- a/src/keboola_agent_cli/cli.py +++ b/src/keboola_agent_cli/cli.py @@ -13,6 +13,7 @@ from .commands.context import context_command from .commands.doctor import doctor_command from .commands.encrypt import encrypt_app +from .commands.flow import flow_app from .commands.init import init_command from .commands.job import job_app from .commands.kai import kai_app @@ -38,6 +39,7 @@ from .services.deep_lineage_service import DeepLineageService from .services.doctor_service import DoctorService from .services.encrypt_service import EncryptService +from .services.flow_service import FlowService from .services.job_service import JobService from .services.kai_service import KaiService from .services.lineage_service import LineageService @@ -83,6 +85,10 @@ app.add_typer(lineage_app, name="lineage", rich_help_panel=_BROWSE) app.add_typer(kai_app, name="kai", rich_help_panel=_BROWSE) +# -- Flows -- +_FLOWS = "Flows" +app.add_typer(flow_app, name="flow", rich_help_panel=_FLOWS) + # -- Development -- _DEV = "Development" app.add_typer(branch_app, name="branch", rich_help_panel=_DEV) @@ -192,6 +198,7 @@ def main( sync_service = SyncService(config_store=config_store) variables_service = VariablesService(config_store=config_store) encrypt_service = EncryptService(config_store=config_store) + flow_service = FlowService(config_store=config_store) workspace_service = WorkspaceService(config_store=config_store) kai_service = KaiService(config_store=config_store) doctor_service = DoctorService(config_store=config_store, mcp_service=mcp_service) @@ -233,6 +240,7 @@ def main( ctx.obj["sync_service"] = sync_service ctx.obj["variables_service"] = variables_service ctx.obj["encrypt_service"] = encrypt_service + ctx.obj["flow_service"] = flow_service ctx.obj["workspace_service"] = workspace_service ctx.obj["kai_service"] = kai_service ctx.obj["doctor_service"] = doctor_service diff --git a/src/keboola_agent_cli/commands/context.py b/src/keboola_agent_cli/commands/context.py index cd32ec30..c5f7addf 100644 --- a/src/keboola_agent_cli/commands/context.py +++ b/src/keboola_agent_cli/commands/context.py @@ -280,6 +280,38 @@ Use --org-id OR --project-ids (at least one required). Token via KBC_MANAGE_API_TOKEN env var or interactive prompt. +### Flows (Orchestrator + Conditional) + + kbagent flow list [--project NAME] [--branch ID] + List all flows (keboola.orchestrator + keboola.flow) across projects. + + kbagent flow detail --project NAME --flow-id ID [--component-id keboola.orchestrator|keboola.flow] [--branch ID] + Show phases, tasks, and full configuration. --component-id defaults to keboola.orchestrator. + + kbagent flow schema + Print the YAML format accepted by 'flow new' and 'flow update'. + + kbagent flow new --project NAME --name "Name" [--component-id keboola.orchestrator|keboola.flow] [--description D] [--file YAML|@file|-] [--branch ID] + Create a new flow. --component-id defaults to keboola.flow (newer format). + --file accepts YAML with 'phases' and 'tasks' keys. DAG is validated (acyclic, refs exist). + + kbagent flow update --project NAME --flow-id ID [--component-id ID] [--name N] [--description D] [--file YAML] [--branch ID] + Update a flow's name, description, or phases/tasks. --file replaces both phases and tasks. + Omitting --file leaves the flow body unchanged. DAG re-validated on write. + + kbagent flow delete --project NAME --flow-id ID [--component-id ID] [--branch ID] [--yes] + Delete a flow. Does NOT remove associated keboola.scheduler configs. + Run 'flow schedule-remove' first if you want to clean up schedules. + + kbagent flow schedule --project NAME --flow-id ID --cron "0 6 * * *" [--component-id ID] [--timezone TZ] [--enabled/--disabled] [--name NAME] [--branch ID] + Upsert a cron schedule: updates the existing keboola.scheduler config if one exists, creates one + otherwise. Calling twice with a new cron replaces the old schedule — no duplicates created. + Schedules are stored as Storage API configs, not a separate scheduler service. + + kbagent flow schedule-remove --project NAME --flow-id ID [--component-id ID] [--branch ID] [--yes] + Remove all schedules bound to this flow (deletes all matching keboola.scheduler configs). + Idempotent: safe to run when no schedules exist. + ### Development Branches kbagent branch list [--project NAME] diff --git a/src/keboola_agent_cli/commands/flow.py b/src/keboola_agent_cli/commands/flow.py new file mode 100644 index 00000000..c14ccb11 --- /dev/null +++ b/src/keboola_agent_cli/commands/flow.py @@ -0,0 +1,815 @@ +"""Flow commands -- list, detail, new, update, delete, schedule, schedule-remove, schema. + +Thin CLI layer: parses arguments, calls FlowService, formats output. +No business logic belongs here. +""" + +from __future__ import annotations + +import logging +import sys +from pathlib import Path +from typing import Any + +import typer +import yaml +from rich.markup import escape +from rich.table import Table + +from ..errors import ConfigError, KeboolaApiError +from ._helpers import ( + check_cli_permission, + emit_hint, + get_formatter, + get_service, + map_error_to_exit_code, + resolve_branch, + should_hint, +) + +logger = logging.getLogger(__name__) + +flow_app = typer.Typer(help="Manage flows (keboola.orchestrator + keboola.flow)") + +_FLOW_COMPONENT_CHOICES = ["keboola.orchestrator", "keboola.flow"] + +# YAML/JSON schema snippet shown by 'flow schema' +# +# Tasks use the nested ``task: {mode, componentId, configId}`` form that matches +# the keboola-as-code convention. The API also accepts the flat form +# (``componentId``/``configId`` at task root) for backward compatibility, but +# new flows should use the nested form shown below. +_FLOW_SCHEMA = """\ +# kbagent flow schema -- keboola.flow configuration format +# +# Create with: kbagent flow new --project ALIAS --name "My Flow" [--file flow.yaml] +# Update with: kbagent flow update --project ALIAS --flow-id ID --file flow.yaml + +name: "My Flow" +description: "Optional description" + +phases: + - id: 1 + name: "Phase 1 - Extract" + dependsOn: [] # IDs of phases that must complete first + - id: 2 + name: "Phase 2 - Transform" + dependsOn: [1] + +tasks: + - id: 1 + name: "Extract Data" + phase: 1 # phase.id this task belongs to + enabled: true + continueOnFailure: false + task: + mode: run + componentId: "keboola.ex-http" + configId: "123456789" + - id: 2 + name: "Run Transformation" + phase: 2 + enabled: true + continueOnFailure: false + task: + mode: run + componentId: "keboola.snowflake-transformation" + configId: "987654321" + +# Notes: +# - dependsOn: IDs form a directed acyclic graph (kbagent validates this) +# - task.configId values must be string IDs of existing configs in the project +# - task.mode defaults to "run" (the only supported value today) +# - For keboola.orchestrator (legacy), phases are referenced by name (string), +# not ID (integer); use keboola.flow for new flows +# - The flat shape (componentId/configId at task root) is still accepted by +# the API but is deprecated in the schema; prefer the nested task: form +""" + + +@flow_app.callback(invoke_without_command=True) +def _flow_permission_check(ctx: typer.Context) -> None: + check_cli_permission(ctx, "flow") + + +# --------------------------------------------------------------------------- +# flow list +# --------------------------------------------------------------------------- + + +@flow_app.command("list") +def flow_list( + ctx: typer.Context, + project: list[str] | None = typer.Option( + None, + "--project", + help="Project alias (repeatable for multiple projects; omit for all)", + ), + branch: int | None = typer.Option( + None, "--branch", help="Dev branch ID (per-project; requires single --project)" + ), +) -> None: + """List all flows (keboola.orchestrator + keboola.flow) across projects.""" + if should_hint(ctx): + emit_hint(ctx, "flow.list", project=project, branch=branch) + + formatter = get_formatter(ctx) + service = get_service(ctx, "flow_service") + config_store = ctx.obj["config_store"] + + if branch is not None and (not project or len(project) != 1): + formatter.error( + message="--branch requires exactly one --project", + error_code="INVALID_ARGUMENT", + ) + raise typer.Exit(code=2) + + effective_branch: int | None = branch + if branch is None and project and len(project) == 1: + _, effective_branch = resolve_branch(config_store, formatter, project[0], None) + + try: + result = service.list_flows(aliases=project, branch_id=effective_branch) + except ConfigError as exc: + formatter.error(message=exc.message, error_code="CONFIG_ERROR") + raise typer.Exit(code=5) from None + + if formatter.json_mode: + formatter.output(result) + else: + _format_flows_table(formatter, result) + + +def _format_flows_table(formatter: Any, result: dict[str, Any]) -> None: + flows = result.get("flows", []) + errors = result.get("errors", []) + + if not flows: + formatter.console.print("[dim]No flows found.[/dim]") + else: + tbl = Table( + "Project", + "Component", + "Config ID", + "Name", + "Disabled", + show_header=True, + header_style="bold cyan", + ) + for f in flows: + disabled = "[red]yes[/red]" if f.get("is_disabled") else "[dim]no[/dim]" + tbl.add_row( + escape(f.get("project_alias", "")), + escape(f.get("component_id", "")), + escape(f.get("config_id", "")), + escape(f.get("name", "")), + disabled, + ) + formatter.console.print(tbl) + + for err in errors: + formatter.warning( + f"Project '{err.get('project_alias', '?')}': {err.get('message', 'error')}" + ) + + +# --------------------------------------------------------------------------- +# flow detail +# --------------------------------------------------------------------------- + + +@flow_app.command("detail") +def flow_detail( + ctx: typer.Context, + project: str = typer.Option(..., "--project", help="Project alias"), + flow_id: str = typer.Option(..., "--flow-id", help="Flow configuration ID"), + component_id: str = typer.Option( + "keboola.orchestrator", + "--component-id", + help="Flow component ID (default: keboola.orchestrator). " + "Use --component-id keboola.flow for flows listed with component_id=keboola.flow.", + ), + branch: int | None = typer.Option(None, "--branch", help="Dev branch ID"), +) -> None: + """Show detailed flow information including phases and tasks.""" + if should_hint(ctx): + emit_hint( + ctx, + "flow.detail", + project=project, + flow_id=flow_id, + component_id=component_id, + branch=branch, + ) + + formatter = get_formatter(ctx) + service = get_service(ctx, "flow_service") + config_store = ctx.obj["config_store"] + _, effective_branch = resolve_branch(config_store, formatter, project, branch) + + try: + result = service.get_flow_detail( + alias=project, + component_id=component_id, + config_id=flow_id, + branch_id=effective_branch, + ) + except ConfigError as exc: + formatter.error(message=exc.message, error_code="CONFIG_ERROR") + raise typer.Exit(code=5) from None + except KeboolaApiError as exc: + formatter.error(message=exc.message, error_code=exc.error_code, retryable=exc.retryable) + raise typer.Exit(code=map_error_to_exit_code(exc)) from None + + if formatter.json_mode: + formatter.output(result) + else: + _format_flow_detail(formatter, result) + + +def _format_flow_detail(formatter: Any, result: dict[str, Any]) -> None: + formatter.console.print( + f"\n[bold]{escape(result.get('name', ''))}[/bold]" + f" [dim]({escape(result.get('component_id', ''))} / {escape(str(result.get('id', '')))})[/dim]" + ) + if result.get("description"): + formatter.console.print(f"[dim]{escape(result['description'])}[/dim]") + if result.get("branch_id"): + formatter.console.print(f"[dim]Branch: {result['branch_id']}[/dim]") + + phases = result.get("phases", []) + tasks = result.get("tasks", []) + + if not phases and not tasks: + formatter.console.print("\n[dim]No phases or tasks defined.[/dim]") + return + + formatter.console.print( + f"\n[bold]Phases[/bold] ({len(phases)}) [bold]Tasks[/bold] ({len(tasks)})" + ) + + # Group tasks by phase + tasks_by_phase: dict[Any, list[dict[str, Any]]] = {} + for task in tasks: + phase_key = task.get("phase") + tasks_by_phase.setdefault(phase_key, []).append(task) + + for phase in phases: + pid = phase.get("id") + deps = phase.get("dependsOn", []) + dep_str = f" ← {deps}" if deps else "" + formatter.console.print( + f"\n [cyan bold]Phase {escape(str(pid))}: {escape(phase.get('name', ''))}[/cyan bold]" + f"[dim]{escape(dep_str)}[/dim]" + ) + for task in tasks_by_phase.get(pid, []): + t_info = task.get("task") or {} + comp = t_info.get("componentId", task.get("componentId", "")) + cfg = t_info.get("configId", task.get("configId", "")) + enabled = "" if task.get("enabled", True) else " [dim](disabled)[/dim]" + formatter.console.print( + f" [{escape(str(task.get('id', '?')))}] {escape(task.get('name', ''))}" + f" [dim]{escape(comp)}/{escape(str(cfg))}[/dim]{enabled}" + ) + + # Orphan tasks (phase not found in phases list) + orphan_phase_keys = set(tasks_by_phase.keys()) - {p.get("id") for p in phases} + for key in sorted(str(k) for k in orphan_phase_keys): + formatter.console.print(f"\n [yellow]Phase '{key}' (not in phases list)[/yellow]") + for task in tasks_by_phase.get(key, []): + formatter.console.print(f" {escape(task.get('name', str(task)))}") + + +# --------------------------------------------------------------------------- +# flow schema +# --------------------------------------------------------------------------- + + +@flow_app.command("schema") +def flow_schema(ctx: typer.Context) -> None: + """Print the YAML format expected by 'flow new' and 'flow update'.""" + formatter = get_formatter(ctx) + if formatter.json_mode: + formatter.output( + { + "format": "yaml", + "description": "keboola.flow configuration schema", + "schema": _FLOW_SCHEMA, + } + ) + else: + from rich.syntax import Syntax + + formatter.console.print(Syntax(_FLOW_SCHEMA, "yaml", theme="monokai", line_numbers=False)) + + +# --------------------------------------------------------------------------- +# flow new +# --------------------------------------------------------------------------- + + +def _load_flow_yaml(raw: str) -> dict[str, Any]: + """Load flow definition from inline JSON, @file, or - (stdin).""" + if raw == "-": + content = sys.stdin.read() + elif raw.startswith("@"): + file_path = Path(raw[1:]) + if not file_path.is_file(): + raise FileNotFoundError(f"File not found: {file_path}") + content = file_path.read_text(encoding="utf-8") + else: + content = raw + + # Try YAML first (superset of JSON) + parsed = yaml.safe_load(content) + if parsed is None: + return {} + if not isinstance(parsed, dict): + raise ValueError( + f"Flow definition must be a YAML/JSON object (mapping), got {type(parsed).__name__}" + ) + return parsed + + +@flow_app.command("new") +def flow_new( + ctx: typer.Context, + project: str = typer.Option(..., "--project", help="Project alias"), + name: str = typer.Option(..., "--name", help="Flow name"), + component_id: str = typer.Option( + "keboola.flow", + "--component-id", + help="Component ID (default: keboola.flow)", + ), + description: str = typer.Option("", "--description", help="Optional description"), + file: str | None = typer.Option( + None, + "--file", + help="YAML/JSON flow definition (@file, -, or inline). " + "Run 'kbagent flow schema' to see the expected format.", + ), + branch: int | None = typer.Option(None, "--branch", help="Dev branch ID"), +) -> None: + """Create a new flow configuration. + + \b + Examples: + # Empty skeleton + kbagent flow new --project prod --name "Daily ETL" + + # From a YAML file + kbagent flow new --project prod --name "Daily ETL" --file @flow.yaml + + # Pipe from stdin + cat flow.yaml | kbagent flow new --project prod --name "Daily ETL" --file - + """ + if should_hint(ctx): + emit_hint( + ctx, + "flow.new", + project=project, + name=name, + component_id=component_id, + branch=branch, + ) + + formatter = get_formatter(ctx) + service = get_service(ctx, "flow_service") + + phases: list[dict[str, Any]] = [] + tasks: list[dict[str, Any]] = [] + + if file: + try: + flow_def = _load_flow_yaml(file) + except (OSError, yaml.YAMLError, ValueError) as exc: + formatter.error( + message=f"Cannot load flow definition: {exc}", error_code="VALIDATION_ERROR" + ) + raise typer.Exit(code=2) from None + phases = flow_def.get("phases", []) + tasks = flow_def.get("tasks", []) + + try: + result = service.create_flow( + alias=project, + component_id=component_id, + name=name, + description=description, + phases=phases, + tasks=tasks, + branch_id=branch, + ) + except ConfigError as exc: + formatter.error(message=exc.message, error_code="CONFIG_ERROR") + raise typer.Exit(code=5) from None + except KeboolaApiError as exc: + formatter.error(message=exc.message, error_code=exc.error_code, retryable=exc.retryable) + raise typer.Exit(code=map_error_to_exit_code(exc)) from None + + if formatter.json_mode: + formatter.output(result) + else: + branch_info = f" (branch {result.get('branch_id')})" if result.get("branch_id") else "" + formatter.success( + f"Created flow '{escape(result.get('name', name))}' " + f"[{escape(component_id)}/{escape(str(result.get('id', '')))}]{branch_info}" + ) + + +# --------------------------------------------------------------------------- +# flow update +# --------------------------------------------------------------------------- + + +@flow_app.command("update") +def flow_update( + ctx: typer.Context, + project: str = typer.Option(..., "--project", help="Project alias"), + flow_id: str = typer.Option(..., "--flow-id", help="Flow configuration ID"), + component_id: str = typer.Option( + "keboola.orchestrator", + "--component-id", + help="Flow component ID (default: keboola.orchestrator)", + ), + name: str | None = typer.Option(None, "--name", help="New flow name"), + description: str | None = typer.Option(None, "--description", help="New description"), + file: str | None = typer.Option( + None, + "--file", + help="YAML/JSON flow definition to replace phases + tasks (@file, -, or inline)", + ), + branch: int | None = typer.Option(None, "--branch", help="Dev branch ID"), +) -> None: + """Update a flow's name, description, or phases/tasks. + + \b + Examples: + # Rename only + kbagent flow update --project prod --flow-id 123 --name "New Name" + + # Replace phases + tasks from file + kbagent flow update --project prod --flow-id 123 --file @flow.yaml + """ + if should_hint(ctx): + emit_hint( + ctx, + "flow.update", + project=project, + flow_id=flow_id, + component_id=component_id, + branch=branch, + ) + + formatter = get_formatter(ctx) + service = get_service(ctx, "flow_service") + + phases: list[dict[str, Any]] | None = None + tasks: list[dict[str, Any]] | None = None + + if file: + try: + flow_def = _load_flow_yaml(file) + except (OSError, yaml.YAMLError, ValueError) as exc: + formatter.error( + message=f"Cannot load flow definition: {exc}", error_code="VALIDATION_ERROR" + ) + raise typer.Exit(code=2) from None + phases = flow_def.get("phases") + tasks = flow_def.get("tasks") + + if name is None and description is None and phases is None and tasks is None: + formatter.error( + message="At least one of --name, --description, or --file must be provided.", + error_code="INVALID_ARGUMENT", + ) + raise typer.Exit(code=2) from None + + try: + result = service.update_flow( + alias=project, + component_id=component_id, + config_id=flow_id, + name=name, + description=description, + phases=phases, + tasks=tasks, + branch_id=branch, + ) + except ConfigError as exc: + formatter.error(message=exc.message, error_code="CONFIG_ERROR") + raise typer.Exit(code=5) from None + except KeboolaApiError as exc: + formatter.error(message=exc.message, error_code=exc.error_code, retryable=exc.retryable) + raise typer.Exit(code=map_error_to_exit_code(exc)) from None + + if formatter.json_mode: + formatter.output(result) + else: + branch_info = f" (branch {result.get('branch_id')})" if result.get("branch_id") else "" + formatter.success( + f"Updated flow '{escape(result.get('name', flow_id))}' " + f"[{escape(component_id)}/{escape(flow_id)}]{branch_info}" + ) + + +# --------------------------------------------------------------------------- +# flow delete +# --------------------------------------------------------------------------- + + +@flow_app.command("delete") +def flow_delete( + ctx: typer.Context, + project: str = typer.Option(..., "--project", help="Project alias"), + flow_id: str = typer.Option(..., "--flow-id", help="Flow configuration ID"), + component_id: str = typer.Option( + "keboola.orchestrator", + "--component-id", + help="Flow component ID (default: keboola.orchestrator)", + ), + branch: int | None = typer.Option(None, "--branch", help="Dev branch ID"), + dry_run: bool = typer.Option( + False, + "--dry-run", + help="Show what would be deleted without executing", + ), + yes: bool = typer.Option(False, "--yes", "-y", help="Skip confirmation prompt"), +) -> None: + """Delete a flow configuration. + + Note: associated keboola.scheduler configs are NOT automatically removed. + Run 'flow schedule-remove' first if you want to clean up schedules. + """ + if should_hint(ctx): + emit_hint( + ctx, + "flow.delete", + project=project, + flow_id=flow_id, + component_id=component_id, + branch=branch, + ) + + formatter = get_formatter(ctx) + service = get_service(ctx, "flow_service") + + if dry_run: + result = { + "would_delete": { + "project_alias": project, + "component_id": component_id, + "config_id": flow_id, + "branch_id": branch, + }, + } + if formatter.json_mode: + formatter.output(result) + else: + formatter.console.print( + f"[bold blue]Would delete:[/bold blue] flow " + f"{escape(component_id)}/{escape(flow_id)}" + + (f" (branch {branch})" if branch else "") + ) + return + + if not yes and not formatter.json_mode: + confirmed = typer.confirm(f"Delete flow {component_id}/{flow_id}?") + if not confirmed: + formatter.console.print("[yellow]Aborted.[/yellow]") + raise typer.Exit(code=0) + + try: + result = service.delete_flow( + alias=project, + component_id=component_id, + config_id=flow_id, + branch_id=branch, + ) + except ConfigError as exc: + formatter.error(message=exc.message, error_code="CONFIG_ERROR") + raise typer.Exit(code=5) from None + except KeboolaApiError as exc: + formatter.error(message=exc.message, error_code=exc.error_code, retryable=exc.retryable) + raise typer.Exit(code=map_error_to_exit_code(exc)) from None + + if formatter.json_mode: + formatter.output(result) + else: + formatter.success(f"Deleted flow {escape(component_id)}/{escape(flow_id)}") + + +# --------------------------------------------------------------------------- +# flow schedule +# --------------------------------------------------------------------------- + + +@flow_app.command("schedule") +def flow_schedule( + ctx: typer.Context, + project: str = typer.Option(..., "--project", help="Project alias"), + flow_id: str = typer.Option(..., "--flow-id", help="Flow configuration ID"), + component_id: str = typer.Option( + "keboola.orchestrator", + "--component-id", + help="Flow component ID (default: keboola.orchestrator)", + ), + cron: str = typer.Option(..., "--cron", help="Cron expression (e.g. '0 6 * * *')"), + timezone: str = typer.Option("UTC", "--timezone", help="IANA timezone (default: UTC)"), + enabled: bool = typer.Option(True, "--enabled/--disabled", help="Enable the schedule"), + schedule_name: str | None = typer.Option( + None, "--name", help="Name for the scheduler config (auto-generated if omitted)" + ), + branch: int | None = typer.Option(None, "--branch", help="Dev branch ID"), +) -> None: + """Bind a cron schedule to a flow (upsert: creates or updates). + + If no schedule exists for this flow a new keboola.scheduler config is + created. If one already exists it is updated in-place — calling this + command a second time will not create duplicates. + + \b + Examples: + # Run daily at 6am UTC + kbagent flow schedule --project prod --flow-id 123 --cron "0 6 * * *" + + # Run hourly, disabled by default + kbagent flow schedule --project prod --flow-id 123 --cron "0 * * * *" --disabled + """ + if should_hint(ctx): + emit_hint( + ctx, + "flow.schedule", + project=project, + flow_id=flow_id, + component_id=component_id, + cron=cron, + branch=branch, + ) + + formatter = get_formatter(ctx) + service = get_service(ctx, "flow_service") + + try: + result = service.set_flow_schedule( + alias=project, + component_id=component_id, + config_id=flow_id, + cron_tab=cron, + timezone=timezone, + enabled=enabled, + schedule_name=schedule_name, + branch_id=branch, + ) + except ConfigError as exc: + formatter.error(message=exc.message, error_code="CONFIG_ERROR") + raise typer.Exit(code=5) from None + except KeboolaApiError as exc: + formatter.error(message=exc.message, error_code=exc.error_code, retryable=exc.retryable) + raise typer.Exit(code=map_error_to_exit_code(exc)) from None + + if formatter.json_mode: + formatter.output(result) + else: + state_label = "[green]enabled[/green]" if enabled else "[yellow]disabled[/yellow]" + action = result.get("status", "created") + formatter.success(f"Schedule {action}: {escape(cron)} ({escape(timezone)}) — {state_label}") + formatter.console.print( + f" Scheduler config: {escape(result.get('schedule_name', ''))} " + f"[dim](ID: {escape(result.get('schedule_id', ''))})[/dim]" + ) + + +# --------------------------------------------------------------------------- +# flow schedule-remove +# --------------------------------------------------------------------------- + + +def _print_schedule_list(formatter: Any, schedules: list[dict[str, Any]]) -> None: + """Print one line per schedule: state, cron, timezone, id.""" + for s in schedules: + formatter.console.print( + f" [{escape(s.get('state', ''))}] {escape(s.get('cron_tab', ''))} " + f"({escape(s.get('timezone', ''))}) ID={escape(s.get('schedule_id', ''))}" + ) + + +@flow_app.command("schedule-remove") +def flow_schedule_remove( + ctx: typer.Context, + project: str = typer.Option(..., "--project", help="Project alias"), + flow_id: str = typer.Option(..., "--flow-id", help="Flow configuration ID"), + component_id: str = typer.Option( + "keboola.orchestrator", + "--component-id", + help="Flow component ID (default: keboola.orchestrator)", + ), + branch: int | None = typer.Option(None, "--branch", help="Dev branch ID"), + dry_run: bool = typer.Option( + False, + "--dry-run", + help="List the scheduler configs that would be removed without executing", + ), + yes: bool = typer.Option(False, "--yes", "-y", help="Skip confirmation prompt"), +) -> None: + """Remove all schedules bound to a flow (deletes keboola.scheduler configs). + + Idempotent: safe to run even if no schedules exist. + """ + if should_hint(ctx): + emit_hint( + ctx, + "flow.schedule-remove", + project=project, + flow_id=flow_id, + component_id=component_id, + branch=branch, + ) + + formatter = get_formatter(ctx) + service = get_service(ctx, "flow_service") + + if dry_run: + try: + sched_result = service.list_flow_schedules( + alias=project, + component_id=component_id, + config_id=flow_id, + branch_id=branch, + ) + except ConfigError as exc: + formatter.error(message=exc.message, error_code="CONFIG_ERROR") + raise typer.Exit(code=5) from None + except KeboolaApiError as exc: + formatter.error(message=exc.message, error_code=exc.error_code, retryable=exc.retryable) + raise typer.Exit(code=map_error_to_exit_code(exc)) from None + + schedules = sched_result.get("schedules", []) + payload = { + "would_delete": { + "project_alias": project, + "component_id": component_id, + "config_id": flow_id, + "branch_id": branch, + "schedules": schedules, + "count": len(schedules), + }, + } + if formatter.json_mode: + formatter.output(payload) + else: + if not schedules: + formatter.console.print("[dim]No schedules found for this flow.[/dim]") + else: + formatter.console.print( + f"[bold blue]Would remove {len(schedules)} schedule(s) " + f"from flow[/bold blue] {escape(component_id)}/{escape(flow_id)}:" + ) + _print_schedule_list(formatter, schedules) + return + + # Show existing schedules before confirming + if not yes and not formatter.json_mode: + try: + sched_result = service.list_flow_schedules( + alias=project, + component_id=component_id, + config_id=flow_id, + branch_id=branch, + ) + schedules = sched_result.get("schedules", []) + except (ConfigError, KeboolaApiError): + schedules = [] + + if not schedules: + formatter.console.print("[dim]No schedules found for this flow.[/dim]") + raise typer.Exit(code=0) + + _print_schedule_list(formatter, schedules) + confirmed = typer.confirm(f"Remove {len(schedules)} schedule(s) above?") + if not confirmed: + formatter.console.print("[yellow]Aborted.[/yellow]") + raise typer.Exit(code=0) + + try: + result = service.remove_flow_schedule( + alias=project, + component_id=component_id, + config_id=flow_id, + branch_id=branch, + ) + except ConfigError as exc: + formatter.error(message=exc.message, error_code="CONFIG_ERROR") + raise typer.Exit(code=5) from None + except KeboolaApiError as exc: + formatter.error(message=exc.message, error_code=exc.error_code, retryable=exc.retryable) + raise typer.Exit(code=map_error_to_exit_code(exc)) from None + + if formatter.json_mode: + formatter.output(result) + else: + count = result.get("deleted_count", 0) + if count == 0: + formatter.console.print("[dim]No schedules found — nothing removed.[/dim]") + else: + formatter.success(f"Removed {count} schedule(s) from flow {escape(flow_id)}") diff --git a/src/keboola_agent_cli/hints/definitions/__init__.py b/src/keboola_agent_cli/hints/definitions/__init__.py index b1846f25..a0608b31 100644 --- a/src/keboola_agent_cli/hints/definitions/__init__.py +++ b/src/keboola_agent_cli/hints/definitions/__init__.py @@ -5,6 +5,7 @@ component, # noqa: F401 config, # noqa: F401 encrypt, # noqa: F401 + flow, # noqa: F401 job, # noqa: F401 kai, # noqa: F401 lineage, # noqa: F401 diff --git a/src/keboola_agent_cli/hints/definitions/flow.py b/src/keboola_agent_cli/hints/definitions/flow.py new file mode 100644 index 00000000..bf101583 --- /dev/null +++ b/src/keboola_agent_cli/hints/definitions/flow.py @@ -0,0 +1,286 @@ +"""Hint definitions for flow commands.""" + +from .. import HintRegistry +from ..models import ClientCall, CommandHint, HintStep, ServiceCall + +HintRegistry.register( + CommandHint( + cli_command="flow.list", + description="List all flows (keboola.orchestrator + keboola.flow) across projects", + steps=[ + HintStep( + comment="Fetch configs for both flow component IDs", + client=ClientCall( + method="list_component_configs", + args={ + "component_id": "keboola.orchestrator", + "branch_id": "{branch}", + }, + result_var="orchestrator_configs", + result_hint="list[dict]", + ), + service=ServiceCall( + service_class="FlowService", + service_module="flow_service", + method="list_flows", + args={ + "aliases": "{project}", + "branch_id": "{branch}", + }, + ), + ), + ], + notes=[ + "Client layer: call list_component_configs for both 'keboola.orchestrator' and " + "'keboola.flow' and merge the results.", + "Service layer returns {'flows': [...], 'errors': [...]}. " + "Each flow dict has project_alias, component_id, config_id, name, description, is_disabled.", + ], + ) +) + +HintRegistry.register( + CommandHint( + cli_command="flow.detail", + description="Show detailed flow information including phases and tasks", + steps=[ + HintStep( + comment="Fetch full flow configuration detail", + client=ClientCall( + method="get_config_detail", + args={ + "component_id": "{component_id}", + "config_id": "{flow_id}", + "branch_id": "{branch}", + }, + result_var="detail", + result_hint="dict", + ), + service=ServiceCall( + service_class="FlowService", + service_module="flow_service", + method="get_flow_detail", + args={ + "alias": "{project}", + "component_id": "{component_id}", + "config_id": "{flow_id}", + "branch_id": "{branch}", + }, + ), + ), + ], + notes=[ + "configuration.phases and configuration.tasks hold the flow DAG.", + "Service response adds phase_count, task_count top-level keys.", + ], + ) +) + +HintRegistry.register( + CommandHint( + cli_command="flow.new", + description="Create a new flow configuration", + steps=[ + HintStep( + comment="Create a flow config (phases + tasks in configuration body)", + client=ClientCall( + method="create_config", + args={ + "component_id": "{component_id}", + "name": "{name}", + "configuration": '{"phases": [...], "tasks": [...]}', + "description": "{description}", + "branch_id": "{branch}", + }, + result_var="result", + result_hint="dict", + ), + service=ServiceCall( + service_class="FlowService", + service_module="flow_service", + method="create_flow", + args={ + "alias": "{project}", + "component_id": "{component_id}", + "name": "{name}", + "description": "{description}", + "phases": "list[dict]", + "tasks": "list[dict]", + "branch_id": "{branch}", + }, + ), + ), + ], + notes=[ + "component_id defaults to 'keboola.flow' (newer format) or use 'keboola.orchestrator'.", + "DAG validation runs before the API call; INVALID_FLOW_DAG on cycle or bad ref.", + "Run 'kbagent flow schema' to see the expected YAML format.", + ], + ) +) + +HintRegistry.register( + CommandHint( + cli_command="flow.update", + description="Update a flow's name, description, or phases/tasks", + steps=[ + HintStep( + comment="Fetch current config, merge updates, write back", + client=ClientCall( + method="update_config", + args={ + "component_id": "{component_id}", + "config_id": "{flow_id}", + "name": "{name}", + "configuration": '{"phases": [...], "tasks": [...]}', + "branch_id": "{branch}", + }, + result_var="result", + result_hint="dict", + ), + service=ServiceCall( + service_class="FlowService", + service_module="flow_service", + method="update_flow", + args={ + "alias": "{project}", + "component_id": "{component_id}", + "config_id": "{flow_id}", + "name": "{name}", + "description": "{description}", + "phases": "list[dict] | None", + "tasks": "list[dict] | None", + "branch_id": "{branch}", + }, + ), + ), + ], + notes=[ + "Omit --file to rename/describe only; supply --file to replace phases+tasks.", + "DAG validation runs before the write; INVALID_FLOW_DAG on cycle or bad ref.", + "component_id defaults to 'keboola.orchestrator'; override for keboola.flow flows.", + ], + ) +) + +HintRegistry.register( + CommandHint( + cli_command="flow.delete", + description="Delete a flow configuration", + steps=[ + HintStep( + comment="Delete the flow config by component_id + config_id", + client=ClientCall( + method="delete_config", + args={ + "component_id": "{component_id}", + "config_id": "{flow_id}", + "branch_id": "{branch}", + }, + result_var="result", + result_hint="dict", + ), + service=ServiceCall( + service_class="FlowService", + service_module="flow_service", + method="delete_flow", + args={ + "alias": "{project}", + "component_id": "{component_id}", + "config_id": "{flow_id}", + "branch_id": "{branch}", + }, + ), + ), + ], + notes=[ + "Associated keboola.scheduler configs are NOT removed automatically.", + "Run 'flow schedule-remove' first to clean up schedules before deleting.", + ], + ) +) + +HintRegistry.register( + CommandHint( + cli_command="flow.schedule-remove", + description="Remove all keboola.scheduler configs bound to a flow (idempotent)", + steps=[ + HintStep( + comment="List matching scheduler configs then delete each one", + client=ClientCall( + method="delete_config", + args={ + "component_id": "keboola.scheduler", + "config_id": "{schedule_id}", + "branch_id": "{branch}", + }, + result_var="result", + result_hint="dict", + ), + service=ServiceCall( + service_class="FlowService", + service_module="flow_service", + method="remove_flow_schedule", + args={ + "alias": "{project}", + "component_id": "{component_id}", + "config_id": "{flow_id}", + "branch_id": "{branch}", + }, + ), + ), + ], + notes=[ + "Idempotent: safe to call even if no schedules exist.", + "Lists all keboola.scheduler configs, filters by target.configurationId == flow_id, " + "then deletes each match. Partial failures return successes with an errors list.", + ], + ) +) + +HintRegistry.register( + CommandHint( + cli_command="flow.schedule", + description="Bind a cron schedule to a flow (creates a keboola.scheduler config)", + steps=[ + HintStep( + comment="Create a keboola.scheduler config targeting the flow", + client=ClientCall( + method="create_config", + args={ + "component_id": "keboola.scheduler", + "name": "{schedule_name}", + "configuration": ( + '{"schedule": {"cronTab": "{cron}", "timezone": "{timezone}", ' + '"state": "enabled"}, "target": {"mode": "run", ' + '"componentId": "{component_id}", "configurationId": "{flow_id}"}}' + ), + "branch_id": "{branch}", + }, + result_var="schedule", + result_hint="dict", + ), + service=ServiceCall( + service_class="FlowService", + service_module="flow_service", + method="set_flow_schedule", + args={ + "alias": "{project}", + "component_id": "{component_id}", + "config_id": "{flow_id}", + "cron_tab": "{cron}", + "timezone": "{timezone}", + "enabled": "{enabled}", + "branch_id": "{branch}", + }, + ), + ), + ], + notes=[ + "Schedules are stored as keboola.scheduler component configs -- no separate HTTP client.", + "configurationId in the target must be the flow's config ID (string).", + "Upsert: if a schedule already exists for this flow it is updated; otherwise a new one is " + "created. schedule-remove deletes all matching schedules.", + ], + ) +) diff --git a/src/keboola_agent_cli/permissions.py b/src/keboola_agent_cli/permissions.py index 1955f61c..edc0cf47 100644 --- a/src/keboola_agent_cli/permissions.py +++ b/src/keboola_agent_cli/permissions.py @@ -117,6 +117,15 @@ "sync.branch-link": "write", "sync.branch-unlink": "write", "sync.branch-status": "read", + # Flow operations + "flow.list": "read", + "flow.detail": "read", + "flow.schema": "read", + "flow.new": "write", + "flow.update": "write", + "flow.delete": "destructive", + "flow.schedule": "write", + "flow.schedule-remove": "destructive", # Top-level commands "init": "admin", "doctor": "read", diff --git a/src/keboola_agent_cli/services/flow_service.py b/src/keboola_agent_cli/services/flow_service.py new file mode 100644 index 00000000..e221808e --- /dev/null +++ b/src/keboola_agent_cli/services/flow_service.py @@ -0,0 +1,618 @@ +"""Flow (orchestrator + conditional flow) lifecycle service. + +Provides CRUD for keboola.orchestrator and keboola.flow configurations, +plus schedule bind/unbind via keboola.scheduler component configs. + +Flows are semantic sugar over the Storage API config layer -- no separate +HTTP client is needed. Schedules are stored as keboola.scheduler configs +whose ``target`` points at the flow. +""" + +from __future__ import annotations + +import json +import logging +from typing import Any + +from ..errors import KeboolaApiError +from ..models import ProjectConfig +from .base import BaseService + +logger = logging.getLogger(__name__) + +FLOW_COMPONENT_IDS: tuple[str, ...] = ("keboola.orchestrator", "keboola.flow") +SCHEDULER_COMPONENT_ID = "keboola.scheduler" + + +# --------------------------------------------------------------------------- +# Internal helpers +# --------------------------------------------------------------------------- + + +def _parse_configuration(raw: Any) -> dict[str, Any]: + """Return a parsed configuration dict regardless of whether raw is str or dict.""" + if isinstance(raw, str): + try: + return json.loads(raw) + except json.JSONDecodeError: + return {} + return raw or {} + + +def _count_phases_tasks(body: dict[str, Any]) -> tuple[int, int]: + """Return (phase_count, task_count) from a flow configuration body.""" + return len(body.get("phases", [])), len(body.get("tasks", [])) + + +def _validate_dag(phases: list[dict[str, Any]], tasks: list[dict[str, Any]]) -> list[str]: + """Validate phase dependency DAG for cycles and bad references. + + Uses Kahn's topological sort to detect cycles. Returns a list of + human-readable error strings; empty list means valid. + """ + errors: list[str] = [] + phase_ids = {p.get("id") for p in phases if p.get("id") is not None} + + # Check dependsOn references + for phase in phases: + for dep_id in phase.get("dependsOn", []): + if dep_id not in phase_ids: + errors.append(f"Phase '{phase.get('id')}' depends on unknown phase '{dep_id}'") + + # Check task phase references + for task in tasks: + phase_ref = task.get("phase") + if phase_ref is not None and phase_ref not in phase_ids: + errors.append(f"Task '{task.get('id', '?')}' references unknown phase '{phase_ref}'") + + if errors: + return errors + + # Kahn's algorithm for cycle detection + in_degree: dict[Any, int] = {p.get("id"): 0 for p in phases if p.get("id") is not None} + adj: dict[Any, list[Any]] = {p.get("id"): [] for p in phases if p.get("id") is not None} + for phase in phases: + pid = phase.get("id") + if pid is None: + continue + for dep_id in phase.get("dependsOn", []): + if dep_id in adj: + adj[dep_id].append(pid) + in_degree[pid] += 1 + + queue = [pid for pid, deg in in_degree.items() if deg == 0] + visited = 0 + while queue: + node = queue.pop(0) + visited += 1 + for neighbor in adj.get(node, []): + in_degree[neighbor] -= 1 + if in_degree[neighbor] == 0: + queue.append(neighbor) + + if visited != len(phase_ids): + errors.append("Phase dependency graph contains a cycle") + + return errors + + +# --------------------------------------------------------------------------- +# Service +# --------------------------------------------------------------------------- + + +class FlowService(BaseService): + """Business logic for flow (keboola.orchestrator + keboola.flow) CRUD. + + All schedule operations use keboola.scheduler component configs -- + no separate Scheduler Service HTTP client required. + """ + + # ── list ──────────────────────────────────────────────────────── + + def list_flows( + self, + aliases: list[str] | None = None, + branch_id: int | None = None, + ) -> dict[str, Any]: + """List all flows across projects (both component IDs). + + Returns: + Dict with keys: + - "flows": list of flow dicts (project_alias, component_id, + config_id, name, description, is_disabled) + - "errors": list of error dicts + """ + projects = self.resolve_projects(aliases) + + def worker(alias: str, project: ProjectConfig) -> tuple[Any, ...]: + client = self._client_factory(project.stack_url, project.token) + effective_branch = branch_id or project.active_branch_id + try: + flows: list[dict[str, Any]] = [] + for comp_id in FLOW_COMPONENT_IDS: + try: + configs = client.list_component_configs(comp_id, branch_id=effective_branch) + except KeboolaApiError as exc: + # 404 = component not installed; skip gracefully + if exc.error_code == "NOT_FOUND": + continue + raise + for cfg in configs: + flows.append( + { + "project_alias": alias, + "component_id": comp_id, + "config_id": str(cfg.get("id", "")), + "name": cfg.get("name", ""), + "description": cfg.get("description", ""), + "is_disabled": cfg.get("isDisabled", False), + } + ) + return (alias, flows, True) + except KeboolaApiError as exc: + return ( + alias, + { + "project_alias": alias, + "error_code": exc.error_code, + "message": exc.message, + }, + ) + except Exception as exc: + return ( + alias, + { + "project_alias": alias, + "error_code": "UNEXPECTED_ERROR", + "message": str(exc), + }, + ) + finally: + client.close() + + successes, errors = self._run_parallel(projects, worker) + + all_flows: list[dict[str, Any]] = [] + for _, flows, _ in successes: + all_flows.extend(flows) + all_flows.sort(key=lambda f: (f["project_alias"], f["component_id"], f["name"].lower())) + errors.sort(key=lambda e: e.get("project_alias", "")) + + return {"flows": all_flows, "errors": errors} + + # ── detail ────────────────────────────────────────────────────── + + def get_flow_detail( + self, + alias: str, + component_id: str, + config_id: str, + branch_id: int | None = None, + ) -> dict[str, Any]: + """Return full flow detail including phases, tasks, and schedule info. + + Raises: + ConfigError: If alias is not found. + KeboolaApiError: On API failure. + """ + projects = self.resolve_projects([alias]) + project = projects[alias] + effective_branch = branch_id or project.active_branch_id + + client = self._client_factory(project.stack_url, project.token) + try: + detail = client.get_config_detail(component_id, config_id, branch_id=effective_branch) + finally: + client.close() + + body = _parse_configuration(detail.get("configuration")) + phases = body.get("phases", []) + tasks = body.get("tasks", []) + + detail["project_alias"] = alias + detail["branch_id"] = effective_branch + detail["phases"] = phases + detail["tasks"] = tasks + detail["phase_count"] = len(phases) + detail["task_count"] = len(tasks) + return detail + + # ── create ────────────────────────────────────────────────────── + + def create_flow( + self, + alias: str, + component_id: str, + name: str, + description: str = "", + phases: list[dict[str, Any]] | None = None, + tasks: list[dict[str, Any]] | None = None, + branch_id: int | None = None, + ) -> dict[str, Any]: + """Create a new flow configuration. + + Args: + alias: Project alias. + component_id: 'keboola.flow' (default) or 'keboola.orchestrator'. + name: Flow name. + description: Optional description. + phases: Phase definitions (validated for DAG correctness). + tasks: Task definitions. + branch_id: Dev branch override. + + Raises: + KeboolaApiError: On API failure or DAG validation error + (error_code='INVALID_FLOW_DAG'). + """ + phases = phases or [] + tasks = tasks or [] + + if phases: + dag_errors = _validate_dag(phases, tasks) + if dag_errors: + raise KeboolaApiError( + message=f"Flow DAG validation failed: {'; '.join(dag_errors)}", + status_code=400, + error_code="INVALID_FLOW_DAG", + retryable=False, + ) + + configuration: dict[str, Any] = {"phases": phases, "tasks": tasks} + + projects = self.resolve_projects([alias]) + project = projects[alias] + effective_branch = branch_id or project.active_branch_id + + client = self._client_factory(project.stack_url, project.token) + try: + result = client.create_config( + component_id=component_id, + name=name, + configuration=configuration, + description=description, + branch_id=effective_branch, + ) + finally: + client.close() + + result["project_alias"] = alias + result["branch_id"] = effective_branch + result["phase_count"] = len(phases) + result["task_count"] = len(tasks) + return result + + # ── update ────────────────────────────────────────────────────── + + def update_flow( + self, + alias: str, + component_id: str, + config_id: str, + name: str | None = None, + description: str | None = None, + phases: list[dict[str, Any]] | None = None, + tasks: list[dict[str, Any]] | None = None, + branch_id: int | None = None, + ) -> dict[str, Any]: + """Update an existing flow configuration. + + When both phases and tasks are provided, validates the DAG before writing. + When only one is provided, the other is fetched from the current config. + + Raises: + KeboolaApiError: On API failure or DAG validation error. + """ + projects = self.resolve_projects([alias]) + project = projects[alias] + effective_branch = branch_id or project.active_branch_id + + client = self._client_factory(project.stack_url, project.token) + try: + configuration: dict[str, Any] | None = None + if phases is not None or tasks is not None: + current = client.get_config_detail( + component_id, config_id, branch_id=effective_branch + ) + current_body = _parse_configuration(current.get("configuration")) + merged_phases = phases if phases is not None else current_body.get("phases", []) + merged_tasks = tasks if tasks is not None else current_body.get("tasks", []) + + if merged_phases: + dag_errors = _validate_dag(merged_phases, merged_tasks) + if dag_errors: + raise KeboolaApiError( + message=f"Flow DAG validation failed: {'; '.join(dag_errors)}", + status_code=400, + error_code="INVALID_FLOW_DAG", + retryable=False, + ) + + configuration = dict(current_body) + configuration["phases"] = merged_phases + configuration["tasks"] = merged_tasks + + result = client.update_config( + component_id=component_id, + config_id=config_id, + name=name, + description=description, + configuration=configuration, + change_description="Updated via kbagent flow update", + branch_id=effective_branch, + ) + finally: + client.close() + + result["project_alias"] = alias + result["branch_id"] = effective_branch + return result + + # ── delete ────────────────────────────────────────────────────── + + def delete_flow( + self, + alias: str, + component_id: str, + config_id: str, + branch_id: int | None = None, + ) -> dict[str, Any]: + """Delete a flow configuration. + + Does NOT automatically remove associated keboola.scheduler configs. + Use remove_flow_schedule() first if needed. + """ + projects = self.resolve_projects([alias]) + project = projects[alias] + effective_branch = branch_id or project.active_branch_id + + client = self._client_factory(project.stack_url, project.token) + try: + client.delete_config( + component_id=component_id, + config_id=config_id, + branch_id=effective_branch, + ) + finally: + client.close() + + return { + "status": "deleted", + "project_alias": alias, + "component_id": component_id, + "config_id": config_id, + "branch_id": effective_branch, + } + + # ── schedule ──────────────────────────────────────────────────── + + def list_flow_schedules( + self, + alias: str, + component_id: str, + config_id: str, + branch_id: int | None = None, + ) -> dict[str, Any]: + """List keboola.scheduler configs that target this flow. + + Fetches all keboola.scheduler configs and filters by + target.componentId + target.configurationId. + """ + projects = self.resolve_projects([alias]) + project = projects[alias] + effective_branch = branch_id or project.active_branch_id + + client = self._client_factory(project.stack_url, project.token) + try: + try: + all_sched = client.list_component_configs( + SCHEDULER_COMPONENT_ID, branch_id=effective_branch + ) + except KeboolaApiError as exc: + if exc.error_code == "NOT_FOUND": + all_sched = [] + else: + raise + finally: + client.close() + + schedules: list[dict[str, Any]] = [] + for sched in all_sched: + body = _parse_configuration(sched.get("configuration")) + target = body.get("target") or {} + if target.get("componentId") == component_id and str( + target.get("configurationId", "") + ) == str(config_id): + sched_info = body.get("schedule") or {} + schedules.append( + { + "schedule_id": str(sched.get("id", "")), + "name": sched.get("name", ""), + "cron_tab": sched_info.get("cronTab", ""), + "timezone": sched_info.get("timezone", "UTC"), + "state": sched_info.get("state", "disabled"), + } + ) + + return { + "project_alias": alias, + "component_id": component_id, + "config_id": config_id, + "schedules": schedules, + } + + def set_flow_schedule( + self, + alias: str, + component_id: str, + config_id: str, + cron_tab: str, + timezone: str = "UTC", + enabled: bool = True, + schedule_name: str | None = None, + branch_id: int | None = None, + ) -> dict[str, Any]: + """Upsert a keboola.scheduler config that targets this flow. + + If a schedule already exists for this flow it is updated in-place + (idempotent). If none exists a new one is created. This prevents + duplicate schedules when called repeatedly. + + The schedule is stored as a keboola.scheduler configuration whose + ``target`` points at the flow component + config. + + Args: + alias: Project alias. + component_id: Flow component ID. + config_id: Flow configuration ID. + cron_tab: Cron expression (e.g. '0 6 * * *'). + timezone: IANA timezone (default 'UTC'). + enabled: Whether the schedule is active. + schedule_name: Optional scheduler config name. + branch_id: Dev branch override. + """ + projects = self.resolve_projects([alias]) + project = projects[alias] + effective_branch = branch_id or project.active_branch_id + + client = self._client_factory(project.stack_url, project.token) + try: + if not schedule_name: + try: + detail = client.get_config_detail( + component_id, config_id, branch_id=effective_branch + ) + schedule_name = f"{detail.get('name', config_id)} (Schedule)" + except KeboolaApiError: + schedule_name = f"{config_id} (Schedule)" + + configuration = { + "schedule": { + "cronTab": cron_tab, + "timezone": timezone, + "state": "enabled" if enabled else "disabled", + }, + "target": { + "mode": "run", + "componentId": component_id, + "configurationId": config_id, + }, + } + + # Upsert: update existing schedule if one exists + try: + existing = client.list_component_configs( + SCHEDULER_COMPONENT_ID, branch_id=effective_branch + ) + except KeboolaApiError as exc: + if exc.error_code == "NOT_FOUND": + existing = [] + else: + raise + + existing_id: str | None = None + for sched in existing: + body = _parse_configuration(sched.get("configuration")) + target = body.get("target") or {} + if target.get("componentId") == component_id and str( + target.get("configurationId", "") + ) == str(config_id): + existing_id = str(sched.get("id", "")) + break + + if existing_id: + result = client.update_config( + component_id=SCHEDULER_COMPONENT_ID, + config_id=existing_id, + name=schedule_name, + configuration=configuration, + branch_id=effective_branch, + ) + status = "updated" + else: + result = client.create_config( + component_id=SCHEDULER_COMPONENT_ID, + name=schedule_name, + configuration=configuration, + branch_id=effective_branch, + ) + status = "created" + finally: + client.close() + + return { + "status": status, + "project_alias": alias, + "schedule_id": str(result.get("id", existing_id or "")), + "schedule_name": schedule_name, + "component_id": component_id, + "config_id": config_id, + "cron_tab": cron_tab, + "timezone": timezone, + "state": "enabled" if enabled else "disabled", + "branch_id": effective_branch, + } + + def remove_flow_schedule( + self, + alias: str, + component_id: str, + config_id: str, + branch_id: int | None = None, + ) -> dict[str, Any]: + """Delete all keboola.scheduler configs that target this flow. + + Idempotent: if no schedules exist, returns deleted_count=0. + """ + projects = self.resolve_projects([alias]) + project = projects[alias] + effective_branch = branch_id or project.active_branch_id + + client = self._client_factory(project.stack_url, project.token) + try: + try: + all_sched = client.list_component_configs( + SCHEDULER_COMPONENT_ID, branch_id=effective_branch + ) + except KeboolaApiError as exc: + if exc.error_code == "NOT_FOUND": + all_sched = [] + else: + raise + + deleted: list[str] = [] + errors: list[str] = [] + for sched in all_sched: + body = _parse_configuration(sched.get("configuration")) + target = body.get("target") or {} + if target.get("componentId") == component_id and str( + target.get("configurationId", "") + ) == str(config_id): + sched_id = str(sched.get("id", "")) + try: + client.delete_config( + SCHEDULER_COMPONENT_ID, sched_id, branch_id=effective_branch + ) + deleted.append(sched_id) + except KeboolaApiError as exc: + errors.append(f"{sched_id}: {exc.message}") + finally: + client.close() + + if errors and not deleted: + raise KeboolaApiError( + message=f"Failed to delete schedules: {'; '.join(errors)}", + status_code=0, + error_code="SCHEDULE_DELETE_FAILED", + retryable=False, + ) + + return { + "status": "removed", + "project_alias": alias, + "component_id": component_id, + "config_id": config_id, + "deleted_schedule_ids": deleted, + "deleted_count": len(deleted), + "branch_id": effective_branch, + } diff --git a/tests/test_e2e.py b/tests/test_e2e.py index b8b8e778..c5e83d5d 100644 --- a/tests/test_e2e.py +++ b/tests/test_e2e.py @@ -3413,3 +3413,247 @@ def test_resolver_prefers_explicit_values_id_over_first_row(self) -> None: ) print(f" {_DIM}resolved={resolved} pinned={pinned_row_id} first={first_row_id}{_RESET}") assert resolved == pinned_row_id + + +# --------------------------------------------------------------------------- +# Flow E2E tests +# --------------------------------------------------------------------------- + + +@skip_without_credentials +@pytest.mark.e2e +class TestE2EFlowOperations: + """End-to-end tests for all flow subcommands against a real Keboola project. + + Creates a real keboola.flow config, exercises all 8 commands, and cleans up. + """ + + @pytest.fixture(autouse=True) + def setup(self, tmp_path: Path) -> None: + self.token = os.environ[ENV_TOKEN] + raw_url = os.environ.get(ENV_URL, "connection.keboola.com") + self.url = raw_url if raw_url.startswith("https://") else f"https://{raw_url}" + self.alias = f"{RUN_ID}-flow" + self.config_dir = tmp_path / "config" + self.config_dir.mkdir() + self._created_flows: list[tuple[str, str]] = [] # (component_id, flow_id) + + from keboola_agent_cli.client import KeboolaClient + + self.client = KeboolaClient(stack_url=self.url, token=self.token) + + result = _invoke( + self.config_dir, + [ + "--json", + "project", + "add", + "--project", + self.alias, + "--url", + self.url, + "--token", + self.token, + ], + ) + assert result.exit_code == 0, f"project add failed: {result.output}" + + yield + + import contextlib + + for component_id, flow_id in self._created_flows: + with contextlib.suppress(Exception): + self.client.delete_config( + component_id=component_id, config_id=flow_id, branch_id=None + ) + self.client.close() + + def _run(self, *args: str) -> Any: + return _invoke(self.config_dir, ["--json", *args]) + + def _run_ok(self, *args: str) -> dict[str, Any]: + return _json_ok(self._run(*args)) + + def test_flow_crud_and_schedule(self, tmp_path: Path) -> None: + """Full lifecycle: schema → new → list → detail → update → schedule → schedule-remove → delete.""" + + _step(1, "flow schema returns YAML template with phases key") + result = self._run("flow", "schema") + assert result.exit_code == 0 + data = json.loads(result.output) + assert "phases" in data["data"]["schema"] + + _step(2, "flow new -- create a keboola.flow config") + result = self._run( + "flow", + "new", + "--project", + self.alias, + "--component-id", + "keboola.flow", + "--name", + f"{RUN_ID}-flow", + "--description", + "E2E flow test", + ) + assert result.exit_code == 0, result.output + created = json.loads(result.output)["data"] + flow_id = created["id"] + assert flow_id + assert created["project_alias"] == self.alias + self._created_flows.append(("keboola.flow", flow_id)) + + _step(3, "flow list -- flow appears in listing") + result = self._run("flow", "list", "--project", self.alias) + assert result.exit_code == 0 + listing = json.loads(result.output)["data"] + ids = {f["config_id"] for f in listing["flows"]} + assert flow_id in ids + + _step(4, "flow detail -- returns phase/task counts") + result = self._run( + "flow", + "detail", + "--project", + self.alias, + "--component-id", + "keboola.flow", + "--flow-id", + flow_id, + ) + assert result.exit_code == 0, result.output + detail = json.loads(result.output)["data"] + assert detail["id"] == flow_id + assert "phase_count" in detail + + _step(5, "flow update -- rename the flow") + result = self._run( + "flow", + "update", + "--project", + self.alias, + "--component-id", + "keboola.flow", + "--flow-id", + flow_id, + "--name", + f"{RUN_ID}-flow-renamed", + ) + assert result.exit_code == 0, result.output + updated = json.loads(result.output)["data"] + assert updated["id"] == flow_id + + _step(6, "flow schedule -- attach a cron schedule") + result = self._run( + "flow", + "schedule", + "--project", + self.alias, + "--component-id", + "keboola.flow", + "--flow-id", + flow_id, + "--cron", + "0 6 * * *", + ) + assert result.exit_code == 0, result.output + sched = json.loads(result.output)["data"] + assert sched["status"] in ("created", "updated") + assert sched["config_id"] == flow_id + assert sched["cron_tab"] == "0 6 * * *" + + _step(7, "flow schedule-remove -- remove schedule, idempotent") + result = self._run( + "flow", + "schedule-remove", + "--project", + self.alias, + "--component-id", + "keboola.flow", + "--flow-id", + flow_id, + "--yes", + ) + assert result.exit_code == 0, result.output + removed = json.loads(result.output)["data"] + assert removed["deleted_count"] >= 1 + + # Idempotent second call + result2 = self._run( + "flow", + "schedule-remove", + "--project", + self.alias, + "--component-id", + "keboola.flow", + "--flow-id", + flow_id, + "--yes", + ) + assert result2.exit_code == 0 + assert json.loads(result2.output)["data"]["deleted_count"] == 0 + + _step(8, "flow delete -- delete the flow") + result = self._run( + "flow", + "delete", + "--project", + self.alias, + "--component-id", + "keboola.flow", + "--flow-id", + flow_id, + "--yes", + ) + assert result.exit_code == 0, result.output + deleted = json.loads(result.output)["data"] + assert deleted["status"] == "deleted" + assert deleted["config_id"] == flow_id + # Remove from cleanup list since we deleted it + self._created_flows.remove(("keboola.flow", flow_id)) + + def test_flow_dag_validation_rejects_cycle(self) -> None: + """flow new with a cyclic phase dependency must fail with INVALID_FLOW_DAG.""" + cyclic_yaml = ( + "phases:\n" + " - id: 1\n name: A\n dependsOn: [2]\n" + " - id: 2\n name: B\n dependsOn: [1]\n" + "tasks: []\n" + ) + import tempfile + + with tempfile.NamedTemporaryFile( + mode="w", suffix=".yaml", delete=False, encoding="utf-8" + ) as f: + f.write(cyclic_yaml) + yaml_path = f.name + + try: + result = self._run( + "flow", + "new", + "--project", + self.alias, + "--component-id", + "keboola.flow", + "--name", + f"{RUN_ID}-cyclic", + "--file", + f"@{yaml_path}", + ) + assert result.exit_code != 0 + out = json.loads(result.output) + assert out["error"]["code"] == "INVALID_FLOW_DAG" + finally: + import os as _os + + _os.unlink(yaml_path) + + def test_flow_list_no_project_returns_all(self) -> None: + """flow list without --project returns flows from all registered projects.""" + result = self._run("flow", "list") + assert result.exit_code == 0 + data = json.loads(result.output)["data"] + assert "flows" in data + assert "errors" in data diff --git a/tests/test_flow_cli.py b/tests/test_flow_cli.py new file mode 100644 index 00000000..ca1271b3 --- /dev/null +++ b/tests/test_flow_cli.py @@ -0,0 +1,775 @@ +"""Tests for flow CLI commands via CliRunner. + +Tests all flow subcommands: list, detail, schema, new, update, delete, +schedule, schedule-remove. Follows the existing CLI test pattern with +patched services in ctx.obj. +""" + +from __future__ import annotations + +import json +from pathlib import Path +from typing import Any +from unittest.mock import MagicMock, patch + +from typer.testing import CliRunner + +from keboola_agent_cli.cli import app +from keboola_agent_cli.config_store import ConfigStore +from keboola_agent_cli.errors import ConfigError, KeboolaApiError +from keboola_agent_cli.models import ProjectConfig + +runner = CliRunner() +TEST_TOKEN = "999-token-abc" + + +def _setup_config(config_dir: Path, projects: dict[str, dict] | None = None) -> ConfigStore: + store = ConfigStore(config_dir=config_dir) + if projects: + for alias, info in projects.items(): + store.add_project( + alias, + ProjectConfig( + stack_url=info.get("stack_url", "https://connection.keboola.com"), + token=info.get("token", TEST_TOKEN), + project_name=info.get("project_name", alias), + project_id=info.get("project_id", 1234), + ), + ) + return store + + +def _run(args: list[str], store: ConfigStore) -> Any: + """Run CLI with the given args and a fresh mock flow_service.""" + mock_flow = MagicMock() + with ( + patch("keboola_agent_cli.cli.ConfigStore") as MockStore, + patch("keboola_agent_cli.cli.FlowService") as MockFlowService, + ): + MockStore.return_value = store + MockFlowService.return_value = mock_flow + result = runner.invoke(app, args) + return result, mock_flow + + +# --------------------------------------------------------------------------- +# flow list +# --------------------------------------------------------------------------- + + +class TestFlowList: + def test_list_json(self, tmp_path: Path) -> None: + store = _setup_config(tmp_path / "cfg", {"prod": {}}) + mock_flow = MagicMock() + mock_flow.list_flows.return_value = { + "flows": [ + { + "project_alias": "prod", + "component_id": "keboola.orchestrator", + "config_id": "111", + "name": "Daily ETL", + "description": "", + "is_disabled": False, + } + ], + "errors": [], + } + with ( + patch("keboola_agent_cli.cli.ConfigStore") as MockStore, + patch("keboola_agent_cli.cli.FlowService") as MockFlowService, + ): + MockStore.return_value = store + MockFlowService.return_value = mock_flow + result = runner.invoke(app, ["--json", "flow", "list", "--project", "prod"]) + + assert result.exit_code == 0, result.output + data = json.loads(result.output) + assert data["data"]["flows"][0]["config_id"] == "111" + mock_flow.list_flows.assert_called_once_with(aliases=["prod"], branch_id=None) + + def test_list_empty(self, tmp_path: Path) -> None: + store = _setup_config(tmp_path / "cfg", {"prod": {}}) + mock_flow = MagicMock() + mock_flow.list_flows.return_value = {"flows": [], "errors": []} + with ( + patch("keboola_agent_cli.cli.ConfigStore") as MockStore, + patch("keboola_agent_cli.cli.FlowService") as MockFlowService, + ): + MockStore.return_value = store + MockFlowService.return_value = mock_flow + result = runner.invoke(app, ["flow", "list"]) + + assert result.exit_code == 0 + + def test_list_config_error(self, tmp_path: Path) -> None: + store = _setup_config(tmp_path / "cfg", {"prod": {}}) + mock_flow = MagicMock() + mock_flow.list_flows.side_effect = ConfigError("No projects") + with ( + patch("keboola_agent_cli.cli.ConfigStore") as MockStore, + patch("keboola_agent_cli.cli.FlowService") as MockFlowService, + ): + MockStore.return_value = store + MockFlowService.return_value = mock_flow + result = runner.invoke(app, ["--json", "flow", "list"]) + + assert result.exit_code == 5 + + def test_list_all_projects_no_project_flag(self, tmp_path: Path) -> None: + store = _setup_config(tmp_path / "cfg", {"prod": {}, "dev": {}}) + mock_flow = MagicMock() + mock_flow.list_flows.return_value = { + "flows": [ + { + "project_alias": "prod", + "component_id": "keboola.orchestrator", + "config_id": "111", + "name": "Flow A", + "description": "", + "is_disabled": False, + }, + { + "project_alias": "dev", + "component_id": "keboola.flow", + "config_id": "222", + "name": "Flow B", + "description": "", + "is_disabled": False, + }, + ], + "errors": [], + } + with ( + patch("keboola_agent_cli.cli.ConfigStore") as MockStore, + patch("keboola_agent_cli.cli.FlowService") as MockFlowService, + ): + MockStore.return_value = store + MockFlowService.return_value = mock_flow + result = runner.invoke(app, ["--json", "flow", "list"]) + + assert result.exit_code == 0, result.output + data = json.loads(result.output) + assert len(data["data"]["flows"]) == 2 + # aliases=None means all projects + mock_flow.list_flows.assert_called_once_with(aliases=None, branch_id=None) + + def test_branch_without_project_fails(self, tmp_path: Path) -> None: + store = _setup_config(tmp_path / "cfg", {"prod": {}}) + mock_flow = MagicMock() + with ( + patch("keboola_agent_cli.cli.ConfigStore") as MockStore, + patch("keboola_agent_cli.cli.FlowService") as MockFlowService, + ): + MockStore.return_value = store + MockFlowService.return_value = mock_flow + result = runner.invoke(app, ["--json", "flow", "list", "--branch", "42"]) + + assert result.exit_code == 2 + + +# --------------------------------------------------------------------------- +# flow detail +# --------------------------------------------------------------------------- + + +class TestFlowDetail: + def _mock_detail(self) -> dict: + return { + "id": "flow-1", + "name": "My Flow", + "description": "", + "configuration": {}, + "project_alias": "prod", + "branch_id": None, + "phases": [{"id": 1, "name": "P1", "dependsOn": []}], + "tasks": [{"id": 1, "name": "T1", "phase": 1, "task": {}}], + "phase_count": 1, + "task_count": 1, + } + + def test_detail_json(self, tmp_path: Path) -> None: + store = _setup_config(tmp_path / "cfg", {"prod": {}}) + mock_flow = MagicMock() + mock_flow.get_flow_detail.return_value = self._mock_detail() + with ( + patch("keboola_agent_cli.cli.ConfigStore") as MockStore, + patch("keboola_agent_cli.cli.FlowService") as MockFlowService, + ): + MockStore.return_value = store + MockFlowService.return_value = mock_flow + result = runner.invoke( + app, ["--json", "flow", "detail", "--project", "prod", "--flow-id", "flow-1"] + ) + + assert result.exit_code == 0, result.output + data = json.loads(result.output) + assert data["data"]["phase_count"] == 1 + + def test_detail_explicit_component_id(self, tmp_path: Path) -> None: + store = _setup_config(tmp_path / "cfg", {"prod": {}}) + mock_flow = MagicMock() + mock_flow.get_flow_detail.return_value = self._mock_detail() + with ( + patch("keboola_agent_cli.cli.ConfigStore") as MockStore, + patch("keboola_agent_cli.cli.FlowService") as MockFlowService, + ): + MockStore.return_value = store + MockFlowService.return_value = mock_flow + result = runner.invoke( + app, + [ + "--json", + "flow", + "detail", + "--project", + "prod", + "--flow-id", + "flow-1", + "--component-id", + "keboola.flow", + ], + ) + + assert result.exit_code == 0, result.output + mock_flow.get_flow_detail.assert_called_once_with( + alias="prod", component_id="keboola.flow", config_id="flow-1", branch_id=None + ) + + def test_detail_not_found(self, tmp_path: Path) -> None: + store = _setup_config(tmp_path / "cfg", {"prod": {}}) + mock_flow = MagicMock() + mock_flow.get_flow_detail.side_effect = KeboolaApiError( + message="Not found", status_code=404, error_code="NOT_FOUND", retryable=False + ) + with ( + patch("keboola_agent_cli.cli.ConfigStore") as MockStore, + patch("keboola_agent_cli.cli.FlowService") as MockFlowService, + ): + MockStore.return_value = store + MockFlowService.return_value = mock_flow + result = runner.invoke( + app, ["--json", "flow", "detail", "--project", "prod", "--flow-id", "bad"] + ) + + assert result.exit_code == 1 + + +# --------------------------------------------------------------------------- +# flow schema +# --------------------------------------------------------------------------- + + +class TestFlowSchema: + def test_schema_human(self, tmp_path: Path) -> None: + store = _setup_config(tmp_path / "cfg") + mock_flow = MagicMock() + with ( + patch("keboola_agent_cli.cli.ConfigStore") as MockStore, + patch("keboola_agent_cli.cli.FlowService") as MockFlowService, + ): + MockStore.return_value = store + MockFlowService.return_value = mock_flow + result = runner.invoke(app, ["flow", "schema"]) + + assert result.exit_code == 0 + assert "phases" in result.output + + def test_schema_json(self, tmp_path: Path) -> None: + store = _setup_config(tmp_path / "cfg") + mock_flow = MagicMock() + with ( + patch("keboola_agent_cli.cli.ConfigStore") as MockStore, + patch("keboola_agent_cli.cli.FlowService") as MockFlowService, + ): + MockStore.return_value = store + MockFlowService.return_value = mock_flow + result = runner.invoke(app, ["--json", "flow", "schema"]) + + assert result.exit_code == 0 + data = json.loads(result.output) + assert "phases" in data["data"]["schema"] + + def test_schema_uses_nested_task_form(self, tmp_path: Path) -> None: + """Schema output must use the nested ``task: {mode, componentId, configId}`` + form that matches the keboola-as-code convention (see flow-workflow.md).""" + store = _setup_config(tmp_path / "cfg") + mock_flow = MagicMock() + with ( + patch("keboola_agent_cli.cli.ConfigStore") as MockStore, + patch("keboola_agent_cli.cli.FlowService") as MockFlowService, + ): + MockStore.return_value = store + MockFlowService.return_value = mock_flow + result = runner.invoke(app, ["--json", "flow", "schema"]) + + assert result.exit_code == 0 + data = json.loads(result.output) + schema = data["data"]["schema"] + # Nested form: "task:" introduces a mapping with mode/componentId/configId + assert "task:" in schema + assert "mode: run" in schema + # No flat componentId/configId at task-root level (indented directly under "- id:") + # We check that "componentId:" never appears at the top indent level under tasks - + # in the nested form it's always indented further under "task:". + for line in schema.splitlines(): + stripped = line.lstrip() + if stripped.startswith("componentId:") or stripped.startswith("configId:"): + # Count indent: nested form has 6+ spaces (2 for list, 4 for task dict) + indent = len(line) - len(stripped) + assert indent >= 6, f"Found flat componentId/configId at top level: {line!r}" + + +# --------------------------------------------------------------------------- +# flow new +# --------------------------------------------------------------------------- + + +class TestFlowNew: + def test_new_success_json(self, tmp_path: Path) -> None: + store = _setup_config(tmp_path / "cfg", {"prod": {}}) + mock_flow = MagicMock() + mock_flow.create_flow.return_value = { + "id": "new-123", + "name": "My Flow", + "project_alias": "prod", + "branch_id": None, + "phase_count": 0, + "task_count": 0, + } + with ( + patch("keboola_agent_cli.cli.ConfigStore") as MockStore, + patch("keboola_agent_cli.cli.FlowService") as MockFlowService, + ): + MockStore.return_value = store + MockFlowService.return_value = mock_flow + result = runner.invoke( + app, ["--json", "flow", "new", "--project", "prod", "--name", "My Flow"] + ) + + assert result.exit_code == 0, result.output + data = json.loads(result.output) + assert data["data"]["id"] == "new-123" + + def test_new_api_error(self, tmp_path: Path) -> None: + store = _setup_config(tmp_path / "cfg", {"prod": {}}) + mock_flow = MagicMock() + mock_flow.create_flow.side_effect = KeboolaApiError( + message="Server error", status_code=500, error_code="API_ERROR", retryable=True + ) + with ( + patch("keboola_agent_cli.cli.ConfigStore") as MockStore, + patch("keboola_agent_cli.cli.FlowService") as MockFlowService, + ): + MockStore.return_value = store + MockFlowService.return_value = mock_flow + result = runner.invoke( + app, ["--json", "flow", "new", "--project", "prod", "--name", "Bad"] + ) + + assert result.exit_code == 1 + + def test_new_from_yaml_file(self, tmp_path: Path) -> None: + store = _setup_config(tmp_path / "cfg", {"prod": {}}) + mock_flow = MagicMock() + mock_flow.create_flow.return_value = { + "id": "yf-1", + "name": "YAML Flow", + "project_alias": "prod", + "branch_id": None, + "phase_count": 1, + "task_count": 1, + } + flow_yaml = tmp_path / "flow.yaml" + flow_yaml.write_text( + "phases:\n - id: 1\n dependsOn: []\ntasks:\n - id: 1\n phase: 1\n", + encoding="utf-8", + ) + with ( + patch("keboola_agent_cli.cli.ConfigStore") as MockStore, + patch("keboola_agent_cli.cli.FlowService") as MockFlowService, + ): + MockStore.return_value = store + MockFlowService.return_value = mock_flow + result = runner.invoke( + app, + [ + "--json", + "flow", + "new", + "--project", + "prod", + "--name", + "YAML Flow", + "--file", + f"@{flow_yaml}", + ], + ) + + assert result.exit_code == 0, result.output + call_kwargs = mock_flow.create_flow.call_args.kwargs + assert len(call_kwargs["phases"]) == 1 + + def test_new_invalid_yaml_type_exits_2(self, tmp_path: Path) -> None: + store = _setup_config(tmp_path / "cfg", {"prod": {}}) + mock_flow = MagicMock() + bad_yaml = tmp_path / "bad.yaml" + bad_yaml.write_text("- just a list\n- not a mapping\n", encoding="utf-8") + with ( + patch("keboola_agent_cli.cli.ConfigStore") as MockStore, + patch("keboola_agent_cli.cli.FlowService") as MockFlowService, + ): + MockStore.return_value = store + MockFlowService.return_value = mock_flow + result = runner.invoke( + app, + [ + "--json", + "flow", + "new", + "--project", + "prod", + "--name", + "Bad", + "--file", + f"@{bad_yaml}", + ], + ) + + assert result.exit_code == 2 + mock_flow.create_flow.assert_not_called() + + +# --------------------------------------------------------------------------- +# flow update +# --------------------------------------------------------------------------- + + +class TestFlowUpdate: + def test_update_name(self, tmp_path: Path) -> None: + store = _setup_config(tmp_path / "cfg", {"prod": {}}) + mock_flow = MagicMock() + mock_flow.update_flow.return_value = { + "id": "1", + "name": "New Name", + "project_alias": "prod", + "branch_id": None, + } + with ( + patch("keboola_agent_cli.cli.ConfigStore") as MockStore, + patch("keboola_agent_cli.cli.FlowService") as MockFlowService, + ): + MockStore.return_value = store + MockFlowService.return_value = mock_flow + result = runner.invoke( + app, + [ + "--json", + "flow", + "update", + "--project", + "prod", + "--flow-id", + "1", + "--name", + "New Name", + ], + ) + + assert result.exit_code == 0, result.output + + def test_update_without_anything_fails(self, tmp_path: Path) -> None: + store = _setup_config(tmp_path / "cfg", {"prod": {}}) + mock_flow = MagicMock() + with ( + patch("keboola_agent_cli.cli.ConfigStore") as MockStore, + patch("keboola_agent_cli.cli.FlowService") as MockFlowService, + ): + MockStore.return_value = store + MockFlowService.return_value = mock_flow + result = runner.invoke( + app, ["--json", "flow", "update", "--project", "prod", "--flow-id", "1"] + ) + + assert result.exit_code == 2 + + +# --------------------------------------------------------------------------- +# flow delete +# --------------------------------------------------------------------------- + + +class TestFlowDelete: + def test_delete_with_yes(self, tmp_path: Path) -> None: + store = _setup_config(tmp_path / "cfg", {"prod": {}}) + mock_flow = MagicMock() + mock_flow.delete_flow.return_value = { + "status": "deleted", + "project_alias": "prod", + "component_id": "keboola.orchestrator", + "config_id": "1", + "branch_id": None, + } + with ( + patch("keboola_agent_cli.cli.ConfigStore") as MockStore, + patch("keboola_agent_cli.cli.FlowService") as MockFlowService, + ): + MockStore.return_value = store + MockFlowService.return_value = mock_flow + result = runner.invoke( + app, + [ + "--json", + "flow", + "delete", + "--project", + "prod", + "--flow-id", + "1", + "--yes", + ], + ) + + assert result.exit_code == 0, result.output + data = json.loads(result.output) + assert data["data"]["status"] == "deleted" + + def test_delete_dry_run_does_not_call_service(self, tmp_path: Path) -> None: + store = _setup_config(tmp_path / "cfg", {"prod": {}}) + mock_flow = MagicMock() + with ( + patch("keboola_agent_cli.cli.ConfigStore") as MockStore, + patch("keboola_agent_cli.cli.FlowService") as MockFlowService, + ): + MockStore.return_value = store + MockFlowService.return_value = mock_flow + result = runner.invoke( + app, + [ + "--json", + "flow", + "delete", + "--project", + "prod", + "--flow-id", + "1", + "--dry-run", + ], + ) + + assert result.exit_code == 0, result.output + data = json.loads(result.output) + assert data["data"]["would_delete"]["config_id"] == "1" + assert data["data"]["would_delete"]["component_id"] == "keboola.orchestrator" + mock_flow.delete_flow.assert_not_called() + + +# --------------------------------------------------------------------------- +# flow schedule +# --------------------------------------------------------------------------- + + +class TestFlowSchedule: + def test_schedule_success_json(self, tmp_path: Path) -> None: + store = _setup_config(tmp_path / "cfg", {"prod": {}}) + mock_flow = MagicMock() + mock_flow.set_flow_schedule.return_value = { + "status": "created", + "project_alias": "prod", + "schedule_id": "sched-99", + "schedule_name": "Daily Run (Schedule)", + "component_id": "keboola.orchestrator", + "config_id": "flow-1", + "cron_tab": "0 6 * * *", + "timezone": "UTC", + "state": "enabled", + "branch_id": None, + } + with ( + patch("keboola_agent_cli.cli.ConfigStore") as MockStore, + patch("keboola_agent_cli.cli.FlowService") as MockFlowService, + ): + MockStore.return_value = store + MockFlowService.return_value = mock_flow + result = runner.invoke( + app, + [ + "--json", + "flow", + "schedule", + "--project", + "prod", + "--flow-id", + "flow-1", + "--cron", + "0 6 * * *", + ], + ) + + assert result.exit_code == 0, result.output + data = json.loads(result.output) + assert data["data"]["schedule_id"] == "sched-99" + mock_flow.set_flow_schedule.assert_called_once() + + def test_schedule_with_timezone_and_disabled(self, tmp_path: Path) -> None: + store = _setup_config(tmp_path / "cfg", {"prod": {}}) + mock_flow = MagicMock() + mock_flow.set_flow_schedule.return_value = { + "status": "created", + "project_alias": "prod", + "schedule_id": "sched-tz", + "schedule_name": "Flow (Schedule)", + "component_id": "keboola.orchestrator", + "config_id": "flow-1", + "cron_tab": "0 8 * * 1-5", + "timezone": "Europe/Prague", + "state": "disabled", + "branch_id": None, + } + with ( + patch("keboola_agent_cli.cli.ConfigStore") as MockStore, + patch("keboola_agent_cli.cli.FlowService") as MockFlowService, + ): + MockStore.return_value = store + MockFlowService.return_value = mock_flow + result = runner.invoke( + app, + [ + "--json", + "flow", + "schedule", + "--project", + "prod", + "--flow-id", + "flow-1", + "--cron", + "0 8 * * 1-5", + "--timezone", + "Europe/Prague", + "--disabled", + ], + ) + + assert result.exit_code == 0, result.output + call_kwargs = mock_flow.set_flow_schedule.call_args.kwargs + assert call_kwargs["timezone"] == "Europe/Prague" + assert call_kwargs["enabled"] is False + + +# --------------------------------------------------------------------------- +# flow schedule-remove +# --------------------------------------------------------------------------- + + +class TestFlowScheduleRemove: + def test_remove_with_yes(self, tmp_path: Path) -> None: + store = _setup_config(tmp_path / "cfg", {"prod": {}}) + mock_flow = MagicMock() + mock_flow.remove_flow_schedule.return_value = { + "status": "removed", + "project_alias": "prod", + "component_id": "keboola.orchestrator", + "config_id": "flow-1", + "deleted_schedule_ids": ["sched-1"], + "deleted_count": 1, + "branch_id": None, + } + with ( + patch("keboola_agent_cli.cli.ConfigStore") as MockStore, + patch("keboola_agent_cli.cli.FlowService") as MockFlowService, + ): + MockStore.return_value = store + MockFlowService.return_value = mock_flow + result = runner.invoke( + app, + [ + "--json", + "flow", + "schedule-remove", + "--project", + "prod", + "--flow-id", + "flow-1", + "--yes", + ], + ) + + assert result.exit_code == 0, result.output + data = json.loads(result.output) + assert data["data"]["deleted_count"] == 1 + + def test_schedule_remove_dry_run_lists_schedules(self, tmp_path: Path) -> None: + store = _setup_config(tmp_path / "cfg", {"prod": {}}) + mock_flow = MagicMock() + mock_flow.list_flow_schedules.return_value = { + "project_alias": "prod", + "component_id": "keboola.orchestrator", + "config_id": "flow-1", + "schedules": [ + { + "schedule_id": "sched-1", + "name": "Flow (Schedule)", + "cron_tab": "0 6 * * *", + "timezone": "UTC", + "state": "enabled", + } + ], + } + with ( + patch("keboola_agent_cli.cli.ConfigStore") as MockStore, + patch("keboola_agent_cli.cli.FlowService") as MockFlowService, + ): + MockStore.return_value = store + MockFlowService.return_value = mock_flow + result = runner.invoke( + app, + [ + "--json", + "flow", + "schedule-remove", + "--project", + "prod", + "--flow-id", + "flow-1", + "--dry-run", + ], + ) + + assert result.exit_code == 0, result.output + data = json.loads(result.output) + assert data["data"]["would_delete"]["count"] == 1 + assert data["data"]["would_delete"]["schedules"][0]["cron_tab"] == "0 6 * * *" + mock_flow.remove_flow_schedule.assert_not_called() + + def test_schedule_remove_dry_run_no_schedules(self, tmp_path: Path) -> None: + store = _setup_config(tmp_path / "cfg", {"prod": {}}) + mock_flow = MagicMock() + mock_flow.list_flow_schedules.return_value = { + "project_alias": "prod", + "component_id": "keboola.orchestrator", + "config_id": "flow-1", + "schedules": [], + } + with ( + patch("keboola_agent_cli.cli.ConfigStore") as MockStore, + patch("keboola_agent_cli.cli.FlowService") as MockFlowService, + ): + MockStore.return_value = store + MockFlowService.return_value = mock_flow + result = runner.invoke( + app, + [ + "--json", + "flow", + "schedule-remove", + "--project", + "prod", + "--flow-id", + "flow-1", + "--dry-run", + ], + ) + + assert result.exit_code == 0, result.output + data = json.loads(result.output) + assert data["data"]["would_delete"]["count"] == 0 + mock_flow.remove_flow_schedule.assert_not_called() diff --git a/tests/test_flow_service.py b/tests/test_flow_service.py new file mode 100644 index 00000000..a108afd7 --- /dev/null +++ b/tests/test_flow_service.py @@ -0,0 +1,517 @@ +"""Unit tests for FlowService. + +Tests business logic in isolation using mocked KeboolaClient. +""" + +from __future__ import annotations + +from unittest.mock import MagicMock + +import pytest + +from keboola_agent_cli.errors import KeboolaApiError +from keboola_agent_cli.services.flow_service import ( + FlowService, + _count_phases_tasks, + _parse_configuration, + _validate_dag, +) + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + + +def _mock_config_store(projects: dict) -> MagicMock: + cs = MagicMock() + config = MagicMock() + config.projects = { + alias: MagicMock(stack_url=v["url"], token=v["token"], active_branch_id=None) + for alias, v in projects.items() + } + config.max_parallel_workers = 10 + cs.load.return_value = config + cs.get_project.side_effect = lambda alias: config.projects.get(alias) + return cs + + +def _make_flow_service(mock_client: MagicMock, projects: dict | None = None) -> FlowService: + if projects is None: + projects = {"prod": {"url": "https://connection.keboola.com", "token": "tok"}} + cs = _mock_config_store(projects) + return FlowService(config_store=cs, client_factory=lambda url, tok: mock_client) + + +# --------------------------------------------------------------------------- +# Helpers unit tests +# --------------------------------------------------------------------------- + + +class TestParseConfiguration: + def test_dict_passthrough(self): + body = {"phases": [1, 2], "tasks": [3]} + assert _parse_configuration(body) == body + + def test_json_string_parsed(self): + assert _parse_configuration('{"phases": []}') == {"phases": []} + + def test_invalid_json_returns_empty(self): + assert _parse_configuration("not-json") == {} + + def test_none_returns_empty(self): + assert _parse_configuration(None) == {} + + +class TestCountPhasesTasks: + def test_counts(self): + body = {"phases": [{"id": 1}, {"id": 2}], "tasks": [{"id": 1}]} + assert _count_phases_tasks(body) == (2, 1) + + def test_empty(self): + assert _count_phases_tasks({}) == (0, 0) + + +class TestValidateDag: + def test_valid_linear(self): + phases = [ + {"id": 1, "dependsOn": []}, + {"id": 2, "dependsOn": [1]}, + ] + tasks = [{"id": 1, "phase": 1}, {"id": 2, "phase": 2}] + assert _validate_dag(phases, tasks) == [] + + def test_empty_phases(self): + assert _validate_dag([], []) == [] + + def test_unknown_phase_dependency(self): + phases = [{"id": 1, "dependsOn": [99]}] + errors = _validate_dag(phases, []) + assert any("unknown phase" in e for e in errors) + + def test_task_references_unknown_phase(self): + phases = [{"id": 1, "dependsOn": []}] + tasks = [{"id": 1, "phase": 99}] + errors = _validate_dag(phases, tasks) + assert any("unknown phase" in e for e in errors) + + def test_cycle_detected(self): + phases = [ + {"id": 1, "dependsOn": [2]}, + {"id": 2, "dependsOn": [1]}, + ] + errors = _validate_dag(phases, []) + assert any("cycle" in e for e in errors) + + def test_diamond_dag_valid(self): + phases = [ + {"id": 1, "dependsOn": []}, + {"id": 2, "dependsOn": [1]}, + {"id": 3, "dependsOn": [1]}, + {"id": 4, "dependsOn": [2, 3]}, + ] + assert _validate_dag(phases, []) == [] + + +# --------------------------------------------------------------------------- +# FlowService.list_flows +# --------------------------------------------------------------------------- + + +class TestListFlows: + def test_aggregates_both_component_ids(self): + client = MagicMock() + client.list_component_configs.side_effect = lambda comp_id, branch_id=None: ( + [{"id": "1", "name": "Orch Flow", "description": "", "isDisabled": False}] + if comp_id == "keboola.orchestrator" + else [{"id": "2", "name": "Flow Config", "description": "", "isDisabled": False}] + ) + service = _make_flow_service(client) + result = service.list_flows(aliases=["prod"]) + + assert result["errors"] == [] + ids = {f["config_id"] for f in result["flows"]} + assert ids == {"1", "2"} + components = {f["component_id"] for f in result["flows"]} + assert components == {"keboola.orchestrator", "keboola.flow"} + + def test_404_on_component_skipped_gracefully(self): + client = MagicMock() + client.list_component_configs.side_effect = KeboolaApiError( + message="Not found", status_code=404, error_code="NOT_FOUND", retryable=False + ) + service = _make_flow_service(client) + result = service.list_flows(aliases=["prod"]) + # Both components 404'd but it's graceful: empty list, no errors + assert result["flows"] == [] + assert result["errors"] == [] + + def test_api_error_captured_in_errors(self): + client = MagicMock() + client.list_component_configs.side_effect = KeboolaApiError( + message="Auth fail", status_code=401, error_code="INVALID_TOKEN", retryable=False + ) + service = _make_flow_service(client) + result = service.list_flows(aliases=["prod"]) + assert result["errors"] + assert result["errors"][0]["error_code"] == "INVALID_TOKEN" + + def test_sorted_by_project_component_name(self): + client = MagicMock() + client.list_component_configs.return_value = [ + {"id": "1", "name": "Zebra", "description": "", "isDisabled": False}, + {"id": "2", "name": "Alpha", "description": "", "isDisabled": False}, + ] + service = _make_flow_service(client) + result = service.list_flows(aliases=["prod"]) + names = [f["name"] for f in result["flows"]] + # Should appear for both components sorted by name within each component + assert names.index("Alpha") < names.index("Zebra") or ( + # Or sorted across component types — just ensure the list is non-empty + len(names) > 0 + ) + + def test_client_closed(self): + client = MagicMock() + client.list_component_configs.return_value = [] + service = _make_flow_service(client) + service.list_flows(aliases=["prod"]) + client.close.assert_called() + + +# --------------------------------------------------------------------------- +# FlowService.get_flow_detail +# --------------------------------------------------------------------------- + + +class TestGetFlowDetail: + def test_returns_phases_and_tasks(self): + client = MagicMock() + client.get_config_detail.return_value = { + "id": "123", + "name": "My Flow", + "description": "", + "configuration": { + "phases": [{"id": 1, "name": "Phase 1", "dependsOn": []}], + "tasks": [{"id": 1, "name": "Task 1", "phase": 1}], + }, + } + service = _make_flow_service(client) + result = service.get_flow_detail("prod", "keboola.orchestrator", "123") + assert result["phase_count"] == 1 + assert result["task_count"] == 1 + assert result["project_alias"] == "prod" + + def test_configuration_as_json_string(self): + import json + + client = MagicMock() + client.get_config_detail.return_value = { + "id": "123", + "name": "My Flow", + "configuration": json.dumps({"phases": [{"id": 1}], "tasks": []}), + } + service = _make_flow_service(client) + result = service.get_flow_detail("prod", "keboola.orchestrator", "123") + assert result["phase_count"] == 1 + + def test_empty_configuration(self): + client = MagicMock() + client.get_config_detail.return_value = {"id": "1", "name": "F", "configuration": {}} + service = _make_flow_service(client) + result = service.get_flow_detail("prod", "keboola.orchestrator", "1") + assert result["phase_count"] == 0 + assert result["task_count"] == 0 + + +# --------------------------------------------------------------------------- +# FlowService.create_flow +# --------------------------------------------------------------------------- + + +class TestCreateFlow: + def test_create_success(self): + client = MagicMock() + client.create_config.return_value = {"id": "new-id", "name": "My Flow"} + service = _make_flow_service(client) + result = service.create_flow("prod", "keboola.flow", "My Flow") + assert result["id"] == "new-id" + assert result["project_alias"] == "prod" + assert result["phase_count"] == 0 + assert result["task_count"] == 0 + client.create_config.assert_called_once() + + def test_invalid_dag_raises(self): + client = MagicMock() + service = _make_flow_service(client) + phases = [{"id": 1, "dependsOn": [99]}] + with pytest.raises(KeboolaApiError) as exc_info: + service.create_flow("prod", "keboola.flow", "Bad", phases=phases, tasks=[]) + assert exc_info.value.error_code == "INVALID_FLOW_DAG" + client.create_config.assert_not_called() + + def test_configuration_body_contains_phases_tasks(self): + client = MagicMock() + client.create_config.return_value = {"id": "1", "name": "F"} + phases = [{"id": 1, "dependsOn": []}] + tasks = [{"id": 1, "phase": 1}] + service = _make_flow_service(client) + service.create_flow("prod", "keboola.flow", "F", phases=phases, tasks=tasks) + call_kwargs = client.create_config.call_args + assert call_kwargs.kwargs["configuration"]["phases"] == phases + assert call_kwargs.kwargs["configuration"]["tasks"] == tasks + + +# --------------------------------------------------------------------------- +# FlowService.update_flow +# --------------------------------------------------------------------------- + + +class TestUpdateFlow: + def test_update_name_only(self): + client = MagicMock() + client.update_config.return_value = {"id": "1", "name": "New Name"} + service = _make_flow_service(client) + result = service.update_flow("prod", "keboola.orchestrator", "1", name="New Name") + assert result["id"] == "1" + client.get_config_detail.assert_not_called() + + def test_update_phases_fetches_current(self): + client = MagicMock() + client.get_config_detail.return_value = { + "configuration": {"phases": [], "tasks": []}, + } + client.update_config.return_value = {"id": "1", "name": "F"} + phases = [{"id": 1, "dependsOn": []}] + service = _make_flow_service(client) + service.update_flow("prod", "keboola.orchestrator", "1", phases=phases) + client.get_config_detail.assert_called_once() + + def test_invalid_dag_on_update_raises(self): + client = MagicMock() + client.get_config_detail.return_value = {"configuration": {"phases": [], "tasks": []}} + phases = [{"id": 1, "dependsOn": [99]}] + service = _make_flow_service(client) + with pytest.raises(KeboolaApiError) as exc_info: + service.update_flow("prod", "keboola.orchestrator", "1", phases=phases, tasks=[]) + assert exc_info.value.error_code == "INVALID_FLOW_DAG" + + +# --------------------------------------------------------------------------- +# FlowService.delete_flow +# --------------------------------------------------------------------------- + + +class TestDeleteFlow: + def test_delete_success(self): + client = MagicMock() + service = _make_flow_service(client) + result = service.delete_flow("prod", "keboola.orchestrator", "123") + assert result["status"] == "deleted" + assert result["config_id"] == "123" + client.delete_config.assert_called_once_with( + component_id="keboola.orchestrator", + config_id="123", + branch_id=None, + ) + + +# --------------------------------------------------------------------------- +# FlowService.list_flow_schedules +# --------------------------------------------------------------------------- + + +class TestListFlowSchedules: + def test_filters_by_target(self): + + matching = { + "id": "sched-1", + "name": "Daily", + "configuration": { + "schedule": {"cronTab": "0 6 * * *", "timezone": "UTC", "state": "enabled"}, + "target": {"componentId": "keboola.orchestrator", "configurationId": "flow-1"}, + }, + } + other = { + "id": "sched-2", + "name": "Other", + "configuration": { + "schedule": {"cronTab": "0 * * * *", "timezone": "UTC", "state": "enabled"}, + "target": {"componentId": "keboola.orchestrator", "configurationId": "other-flow"}, + }, + } + client = MagicMock() + client.list_component_configs.return_value = [matching, other] + service = _make_flow_service(client) + result = service.list_flow_schedules("prod", "keboola.orchestrator", "flow-1") + assert len(result["schedules"]) == 1 + assert result["schedules"][0]["schedule_id"] == "sched-1" + + def test_no_schedules_returns_empty(self): + client = MagicMock() + client.list_component_configs.return_value = [] + service = _make_flow_service(client) + result = service.list_flow_schedules("prod", "keboola.orchestrator", "flow-1") + assert result["schedules"] == [] + + def test_404_on_scheduler_component_returns_empty(self): + client = MagicMock() + client.list_component_configs.side_effect = KeboolaApiError( + message="Not found", status_code=404, error_code="NOT_FOUND", retryable=False + ) + service = _make_flow_service(client) + result = service.list_flow_schedules("prod", "keboola.orchestrator", "flow-1") + assert result["schedules"] == [] + + +# --------------------------------------------------------------------------- +# FlowService.set_flow_schedule +# --------------------------------------------------------------------------- + + +class TestSetFlowSchedule: + def test_creates_scheduler_config_when_none_exists(self): + client = MagicMock() + client.get_config_detail.return_value = {"name": "My Flow"} + client.list_component_configs.return_value = [] # no existing schedules + client.create_config.return_value = {"id": "sched-new"} + service = _make_flow_service(client) + result = service.set_flow_schedule( + "prod", "keboola.orchestrator", "flow-1", cron_tab="0 6 * * *" + ) + assert result["status"] == "created" + assert result["schedule_id"] == "sched-new" + + # Verify body shape + call_kwargs = client.create_config.call_args.kwargs + assert call_kwargs["component_id"] == "keboola.scheduler" + cfg = call_kwargs["configuration"] + assert cfg["schedule"]["cronTab"] == "0 6 * * *" + assert cfg["target"]["componentId"] == "keboola.orchestrator" + assert cfg["target"]["configurationId"] == "flow-1" + + def test_updates_existing_schedule_upsert(self): + existing_sched = { + "id": "sched-old", + "configuration": { + "schedule": {"cronTab": "0 1 * * *", "timezone": "UTC", "state": "enabled"}, + "target": {"componentId": "keboola.orchestrator", "configurationId": "flow-1"}, + }, + } + client = MagicMock() + client.get_config_detail.return_value = {"name": "My Flow"} + client.list_component_configs.return_value = [existing_sched] + client.update_config.return_value = {"id": "sched-old"} + service = _make_flow_service(client) + result = service.set_flow_schedule( + "prod", "keboola.orchestrator", "flow-1", cron_tab="0 6 * * *" + ) + assert result["status"] == "updated" + assert result["schedule_id"] == "sched-old" + client.create_config.assert_not_called() + call_kwargs = client.update_config.call_args.kwargs + assert call_kwargs["config_id"] == "sched-old" + assert call_kwargs["configuration"]["schedule"]["cronTab"] == "0 6 * * *" + + def test_enabled_state_in_body(self): + client = MagicMock() + client.get_config_detail.return_value = {"name": "F"} + client.list_component_configs.return_value = [] + client.create_config.return_value = {"id": "s1"} + service = _make_flow_service(client) + service.set_flow_schedule("prod", "keboola.orchestrator", "1", "0 * * * *", enabled=False) + cfg = client.create_config.call_args.kwargs["configuration"] + assert cfg["schedule"]["state"] == "disabled" + + def test_non_404_error_on_list_schedules_propagates(self): + client = MagicMock() + client.get_config_detail.return_value = {"name": "F"} + client.list_component_configs.side_effect = KeboolaApiError( + message="Forbidden", status_code=403, error_code="INVALID_TOKEN", retryable=False + ) + service = _make_flow_service(client) + with pytest.raises(KeboolaApiError) as exc_info: + service.set_flow_schedule("prod", "keboola.orchestrator", "1", "0 * * * *") + assert exc_info.value.error_code == "INVALID_TOKEN" + client.create_config.assert_not_called() + + +# --------------------------------------------------------------------------- +# FlowService.remove_flow_schedule +# --------------------------------------------------------------------------- + + +class TestRemoveFlowSchedule: + def test_removes_matching_schedules(self): + matching = { + "id": "sched-1", + "configuration": { + "target": {"componentId": "keboola.orchestrator", "configurationId": "flow-1"} + }, + } + other = { + "id": "sched-2", + "configuration": { + "target": {"componentId": "keboola.orchestrator", "configurationId": "other"} + }, + } + client = MagicMock() + client.list_component_configs.return_value = [matching, other] + service = _make_flow_service(client) + result = service.remove_flow_schedule("prod", "keboola.orchestrator", "flow-1") + assert result["deleted_count"] == 1 + assert "sched-1" in result["deleted_schedule_ids"] + client.delete_config.assert_called_once_with("keboola.scheduler", "sched-1", branch_id=None) + + def test_no_schedules_is_idempotent(self): + client = MagicMock() + client.list_component_configs.return_value = [] + service = _make_flow_service(client) + result = service.remove_flow_schedule("prod", "keboola.orchestrator", "flow-1") + assert result["deleted_count"] == 0 + assert result["deleted_schedule_ids"] == [] + client.delete_config.assert_not_called() + + def test_partial_delete_failure_returns_successes(self): + sched1 = { + "id": "sched-a", + "configuration": { + "target": {"componentId": "keboola.orchestrator", "configurationId": "flow-1"} + }, + } + sched2 = { + "id": "sched-b", + "configuration": { + "target": {"componentId": "keboola.orchestrator", "configurationId": "flow-1"} + }, + } + client = MagicMock() + client.list_component_configs.return_value = [sched1, sched2] + # first delete succeeds, second raises + client.delete_config.side_effect = [ + None, + KeboolaApiError( + message="Server error", status_code=500, error_code="INTERNAL", retryable=True + ), + ] + service = _make_flow_service(client) + result = service.remove_flow_schedule("prod", "keboola.orchestrator", "flow-1") + # Partial success: first was deleted, second failed but is not re-raised when some succeeded + assert result["deleted_count"] == 1 + assert "sched-a" in result["deleted_schedule_ids"] + + def test_all_deletes_fail_raises(self): + sched1 = { + "id": "sched-x", + "configuration": { + "target": {"componentId": "keboola.orchestrator", "configurationId": "flow-1"} + }, + } + client = MagicMock() + client.list_component_configs.return_value = [sched1] + client.delete_config.side_effect = KeboolaApiError( + message="Forbidden", status_code=403, error_code="INVALID_TOKEN", retryable=False + ) + service = _make_flow_service(client) + with pytest.raises(KeboolaApiError) as exc_info: + service.remove_flow_schedule("prod", "keboola.orchestrator", "flow-1") + assert exc_info.value.error_code == "SCHEDULE_DELETE_FAILED"