host.repeat¶

Periodic background task runner for UnixHost.

RepeatRunner owns the repeat-task lifecycle: starting named periodic coroutines, storing their results in a bounded deque, and cancelling them cleanly.

It is decoupled from SSH/telnet — it receives a run_cmds coroutine factory so it can be tested with an AsyncMock without any real connection.

class otto.host.repeat.RepeatRunner(run_cmds: Callable[[...], Any])¶

Bases: object

Runs named periodic command tasks and stores their results.

Parameters¶

run_cmds:

Async callable that accepts a list[str] | str and returns a RunResult. Typically bound to Host.run.

start(name: str, cmds: list[str] | str, interval: timedelta, times: int = -1, duration: timedelta = datetime.timedelta(days=999999999, seconds=86399, microseconds=999999), until: datetime = datetime.datetime(9999, 12, 31, 23, 59, 59, 999999), on_result: Callable[[str, datetime, list[CommandStatus]], None] | None = None, max_history: int = 1000) → None¶

Start a named periodic task. Raises if that name is already running.

async stop(name: str) → None¶

Cancel a specific named periodic task.

async stop_all() → None¶

Cancel all running periodic tasks.

get_results(name: str) → list[tuple[datetime, list[CommandStatus]]]¶

Return a snapshot of stored results for a named repeat task.