Something went wrong. Try again.
Command line interface for accessing the Open Web Index shards and slices.
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286"""Workflow Commands - Typer CLI for Airflow workflow monitoring.
Commands:- dags: list the DAGs the Airflow instance exposes- runs: DAG runs in a window, with state- executions: individual task executions, filterable- summary: execution statistics by operator, DAG and state
The filtering and the Airflow calls live in `owilix.core.tasks.workflow`, sharedwith the HTTP API so the two cannot drift."""import jsonfrom typing import Optional
import typer
from ._common.context import CLIContext
app = typer.Typer( name="workflow", help="Airflow workflow monitoring", no_args_is_help=True,)
def _emit(cli_ctx: CLIContext, result, as_json: bool) -> None: """Print JSON and exit non-zero on failure, uniformly for every subcommand.""" if not result.success: if as_json: cli_ctx.console.print_json(json.dumps(result.as_dict(), default=str)) else: cli_ctx.console.print(f"[red]{result.msg}[/red]") raise typer.Exit(code=result.exit_code or 1) if as_json: cli_ctx.console.print_json(json.dumps(result.object, default=str))
def _state_colour(state: str) -> str: return { "success": "green", "failed": "red", "running": "yellow", "queued": "cyan", "skipped": "dim", "up_for_reschedule": "yellow", }.get(str(state), "white")
@app.command("dags")def dags( ctx: typer.Context, all_dags: bool = typer.Option(False, "--all", "-a", help="Include paused DAGs"), as_json: bool = typer.Option(False, "--json", help="Emit JSON"), base_url: Optional[str] = typer.Option(None, "--base-url", help="Override the Airflow base URL"),): """List the DAGs available on the Airflow instance.""" from owilix.core.tasks.workflow import workflow_dags
cli_ctx: CLIContext = ctx.obj result = workflow_dags(cli_ctx.owi, base_url=base_url, include_paused=all_dags, console=cli_ctx.console) _emit(cli_ctx, result, as_json) if as_json: return
from rich.table import Table
table = Table(title=f"Airflow DAGs ({result.object['count']})") table.add_column("DAG", style="cyan") table.add_column("Paused") table.add_column("Tags") for dag in result.object["dags"]: table.add_row( str(dag["dag_id"]), "[yellow]yes[/yellow]" if dag["is_paused"] else "[green]no[/green]", ", ".join(str(t) for t in dag["tags"]), ) cli_ctx.console.print(table)
@app.command("runs")def runs( ctx: typer.Context, dag_id: Optional[str] = typer.Option(None, "--dag", "-d", help="DAG id(s), comma-separated"), state: Optional[str] = typer.Option(None, "--state", help="Run state(s), comma-separated"), since: Optional[str] = typer.Option(None, "--since", help="Start of window (YYYY-MM-DD or ISO-8601)"), until: Optional[str] = typer.Option(None, "--until", help="End of window (YYYY-MM-DD or ISO-8601)"), days: Optional[int] = typer.Option(None, "--days", help="Look back this many days (default 7)"), hours: Optional[int] = typer.Option(None, "--hours", help="Look back this many hours"), all_dags: bool = typer.Option(False, "--all", "-a", help="Include paused DAGs"), as_json: bool = typer.Option(False, "--json", help="Emit JSON"), base_url: Optional[str] = typer.Option(None, "--base-url", help="Override the Airflow base URL"),): """List DAG runs and their state.""" from owilix.core.tasks.workflow import workflow_runs
cli_ctx: CLIContext = ctx.obj result = workflow_runs( cli_ctx.owi, base_url=base_url, since=since, until=until, days=days, hours=hours, dag_id=dag_id, state=state, include_paused=all_dags, console=cli_ctx.console, ) _emit(cli_ctx, result, as_json) if as_json: return
from rich.table import Table
data = result.object table = Table(title=f"DAG runs ({data['count']})") table.add_column("DAG", style="cyan") table.add_column("Run") table.add_column("Type") table.add_column("State") table.add_column("Started") for run in data["runs"]: table.add_row( str(run["dag_id"]), str(run.get("dag_run_id") or "")[:44], str(run.get("run_type") or ""), f"[{_state_colour(run.get('state'))}]{run.get('state')}[/]", str(run.get("start_date") or "")[:19], ) cli_ctx.console.print(table) if data["partial"]: cli_ctx.console.print( f"[yellow]Incomplete: {len(data['failed'])} DAG(s) could not be read " f"({', '.join(data['failed'])}).[/yellow]" )
@app.command("executions")def executions( ctx: typer.Context, dag_id: Optional[str] = typer.Option(None, "--dag", "-d", help="DAG id(s), comma-separated"), task_id: Optional[str] = typer.Option(None, "--task", "-t", help="Task id(s), comma-separated"), operator: Optional[str] = typer.Option(None, "--operator", help="Operator class(es), comma-separated"), state: Optional[str] = typer.Option(None, "--state", "-s", help="State(s): success, failed, skipped, queued, up_for_reschedule"), pool: Optional[str] = typer.Option(None, "--pool", help="Airflow pool(s)"), queue: Optional[str] = typer.Option(None, "--queue", help="Queue(s)"), hostname: Optional[str] = typer.Option(None, "--hostname", help="Worker hostname(s)"), min_duration: Optional[float] = typer.Option(None, "--min-duration", help="Only executions at least this many seconds"), since: Optional[str] = typer.Option(None, "--since", help="Start of window (YYYY-MM-DD or ISO-8601)"), until: Optional[str] = typer.Option(None, "--until", help="End of window (YYYY-MM-DD or ISO-8601)"), days: Optional[int] = typer.Option(None, "--days", help="Look back this many days (default 7)"), hours: Optional[int] = typer.Option(None, "--hours", help="Look back this many hours"), limit: Optional[int] = typer.Option(None, "--limit", "-n", help="Show at most this many"), offset: int = typer.Option(0, "--offset", help="Skip this many"), as_json: bool = typer.Option(False, "--json", help="Emit JSON"), base_url: Optional[str] = typer.Option(None, "--base-url", help="Override the Airflow base URL"),): r"""List individual task executions.
One execution is one task, in one DAG run: a state, a duration and the operator that ran it.
Filters combine as OR within a field and AND across fields, so `--state failed,up_for_reschedule --dag X` means "either of those states, in DAG X".
Examples: owi workflow executions --days 7 owi workflow executions --state failed --days 30 owi workflow executions --since 2026-08-01 --until 2026-08-05 --json owi workflow executions --operator LexisAAITokenKeeperSensor --state failed owi workflow executions --min-duration 60 --limit 20 """ from owilix.core.tasks.workflow import workflow_executions
cli_ctx: CLIContext = ctx.obj result = workflow_executions( cli_ctx.owi, base_url=base_url, since=since, until=until, days=days, hours=hours, dag_id=dag_id, task_id=task_id, operator=operator, state=state, pool=pool, queue=queue, hostname=hostname, min_duration=min_duration, limit=limit, offset=offset, console=cli_ctx.console, ) _emit(cli_ctx, result, as_json) if as_json: return
from rich.table import Table
data = result.object window = data["window"] title = f"Executions ({data['count']} of {data['total']})" if window["since"]: title += f" since {window['since'][:19]}" table = Table(title=title) table.add_column("Started") table.add_column("DAG", style="cyan") table.add_column("Task") table.add_column("Operator") table.add_column("State") table.add_column("Sec", justify="right") for item in data["executions"]: duration = item.get("duration") table.add_row( str(item.get("start_date") or "")[:19], str(item.get("dag_id") or "")[:34], str(item.get("task_id") or "")[:26], str(item.get("operator") or "")[:30], f"[{_state_colour(item.get('state'))}]{item.get('state')}[/]", "-" if duration is None else f"{float(duration):.2f}", ) cli_ctx.console.print(table) if data["has_more"]: cli_ctx.console.print( f"[dim]{data['total'] - data['offset'] - data['count']} more; " f"use --offset/--limit to page.[/dim]" )
@app.command("summary")def summary( ctx: typer.Context, dag_id: Optional[str] = typer.Option(None, "--dag", "-d", help="DAG id(s), comma-separated"), task_id: Optional[str] = typer.Option(None, "--task", "-t", help="Task id(s), comma-separated"), operator: Optional[str] = typer.Option(None, "--operator", help="Operator class(es)"), state: Optional[str] = typer.Option(None, "--state", "-s", help="State(s)"), pool: Optional[str] = typer.Option(None, "--pool", help="Airflow pool(s)"), queue: Optional[str] = typer.Option(None, "--queue", help="Queue(s)"), hostname: Optional[str] = typer.Option(None, "--hostname", help="Worker hostname(s)"), min_duration: Optional[float] = typer.Option(None, "--min-duration", help="Only executions at least this many seconds"), since: Optional[str] = typer.Option(None, "--since", help="Start of window (YYYY-MM-DD or ISO-8601)"), until: Optional[str] = typer.Option(None, "--until", help="End of window (YYYY-MM-DD or ISO-8601)"), days: Optional[int] = typer.Option(None, "--days", help="Look back this many days (default 7)"), hours: Optional[int] = typer.Option(None, "--hours", help="Look back this many hours"), as_json: bool = typer.Option(False, "--json", help="Emit JSON"), base_url: Optional[str] = typer.Option(None, "--base-url", help="Override the Airflow base URL"),): """Summarize executions by state, operator and DAG.
Accepts the same filters as `executions`, so a summary always describes exactly the set you just listed.
Examples: owi workflow summary --days 7 owi workflow summary --dag openwebsearch_OpenWebIndex_Main_V2_IT4I --json owi workflow summary --since 2026-08-01 --state failed """ from owilix.core.tasks.workflow import workflow_summary
cli_ctx: CLIContext = ctx.obj result = workflow_summary( cli_ctx.owi, base_url=base_url, since=since, until=until, days=days, hours=hours, dag_id=dag_id, task_id=task_id, operator=operator, state=state, pool=pool, queue=queue, hostname=hostname, min_duration=min_duration, console=cli_ctx.console, ) _emit(cli_ctx, result, as_json) if as_json: return
from rich.table import Table
data = result.object window = data["window"] header = f"[bold]{data['total_executions']:,} execution(s)[/bold]" if window["since"]: header += f" since {window['since'][:19]}" if data["failed"]: header += f" — [red]{data['failed']} failed[/red]" cli_ctx.console.print(header)
states = Table(title="By state") states.add_column("State") states.add_column("Count", justify="right") for name, count in data["states"].items(): states.add_row(f"[{_state_colour(name)}]{name}[/]", f"{count:,}") cli_ctx.console.print(states)
for label, key in (("By operator", "by_operator"), ("By DAG", "by_dag")): table = Table(title=label) table.add_column("Name", style="cyan") table.add_column("Total", justify="right") table.add_column("States") table.add_column("Avg s", justify="right") table.add_column("p95 s", justify="right") for name, entry in data[key].items(): state_text = ", ".join( f"[{_state_colour(s)}]{s}={c}[/]" for s, c in sorted(entry["states"].items()) ) avg = entry["avg_duration_sec"] p95 = entry["p95_duration_sec"] table.add_row( name[:46], f"{entry['total']:,}", state_text, "-" if avg is None else f"{avg:.2f}", "-" if p95 is None else f"{p95:.2f}", ) cli_ctx.console.print(table)