Skip to content

routines_cmd

routines_cmd

diapason routines — cron routine catalog.

Functions:

routines 

routines() -> None

Manage ROUTINES.json cron automations.

Source code in src/diapason/cli/routines_cmd.py 
30
31
32
@click.group("routines")
def routines() -> None:
    """Manage ROUTINES.json cron automations."""

routines_list 

routines_list() -> None

List routines and enabled state.

Source code in src/diapason/cli/routines_cmd.py 
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
@routines.command("list")
def routines_list() -> None:
    """List routines and enabled state."""
    from diapason.heartbeat.routines import load_routines, load_state

    console = Console()
    items = load_routines(_workspace())
    state = load_state(_workspace())
    last = state.get("last_run") or {}
    table = Table(title="Routines")
    table.add_column("ID")
    table.add_column("Kind")
    table.add_column("Cron")
    table.add_column("On")
    table.add_column("Last")
    for r in items:
        lr = last.get(r.id) or {}
        table.add_row(
            r.id,
            r.kind,
            r.cron,
            "yes" if r.enabled else "no",
            str(lr.get("at") or "—")[:19],
        )
    console.print(table)

routines_run 

routines_run(
    routine_id: str, force: bool, dry_run: bool
) -> None

Run a routine once now.

Source code in src/diapason/cli/routines_cmd.py 
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
@routines.command("run")
@click.argument("routine_id")
@click.option("--force", is_flag=True, help="Run even if disabled / quiet hours.")
@click.option("--dry-run", is_flag=True, help="No agent — use dry handlers.")
def routines_run(routine_id: str, force: bool, dry_run: bool) -> None:
    """Run a routine once now."""
    from diapason.heartbeat.kinds import run_routine
    from diapason.heartbeat.routines import get_routine

    console = Console()
    routine = get_routine(routine_id, _workspace())
    if routine is None:
        console.print(f"[red]Unknown routine:[/red] {routine_id}")
        raise SystemExit(1)

    system = None
    if not dry_run:
        try:
            from diapason.sdk import Diapason

            with Diapason() as j:
                out = run_routine(
                    routine, system=j, force=force, workspace=_workspace()
                )
            _print_run(console, routine_id, out)
            return
        except Exception as exc:
            console.print(f"[yellow]Diapason unavailable ({exc}); dry run.[/yellow]")

    out = run_routine(routine, system=system, force=force, workspace=_workspace())
    _print_run(console, routine_id, out)

routines_sync 

routines_sync() -> None

Push heartbeat + routines into the TaskScheduler DB.

Source code in src/diapason/cli/routines_cmd.py 
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
@routines.command("sync")
def routines_sync() -> None:
    """Push heartbeat + routines into the TaskScheduler DB."""
    from diapason.heartbeat.sync import sync_heartbeat_and_routines

    console = Console()
    store, sched = _scheduler()
    try:
        summary = sync_heartbeat_and_routines(sched, workspace=_workspace())
        console.print(f"[green]Synced[/green] heartbeat={summary.get('heartbeat')}")
        for r in summary.get("routines") or []:
            flag = "on" if r.get("active") else "off"
            console.print(f"  {r['task_id']} [{flag}]")
        console.print("[dim]Start the daemon with: diapason scheduler start[/dim]")
    finally:
        store.close()

routines_preview 

routines_preview() -> None

Show which routines would fire (enabled + cron).

Source code in src/diapason/cli/routines_cmd.py 
152
153
154
155
156
157
158
159
160
@routines.command("preview")
def routines_preview() -> None:
    """Show which routines would fire (enabled + cron)."""
    from diapason.heartbeat.routines import load_routines

    console = Console()
    for r in load_routines(_workspace()):
        status = "would-fire" if r.enabled else "disabled"
        console.print(f"{r.id:20} {r.cron:20} {status}")