context¶

otto’s per-invocation runtime composition root.

Owns the active Lab, the per-invocation runtime flags, and the host lifecycle scope. Propagated via a ContextVar so the bare module accessors (otto.configmodule.all_hosts/get_host) can stay zero-argument, while explicit passing (OttoContext methods, open_context) is first-class.

class otto.context.HostScope¶

Bases: object

Owns hosts handed out during a command; closes any still-connected on exit.

The deterministic backstop that replaces RemoteHost.__del__: a host created and passed around without an explicit async with is still closed when the scope exits. Registration is deduped by object identity; close() is assumed idempotent so an early per-host close and the sweep never collide.

register(host: RemoteHost) → None¶

otto.context.get_context() → OttoContext¶

otto.context.try_get_context() → OttoContext | None¶

otto.context.set_context(ctx: OttoContext) → Token[OttoContext | None]¶

otto.context.reset_context(token: Token[OttoContext | None]) → None¶

class otto.context.OttoContext(lab: Lab, dry_run: bool = False, log_command_output: bool = True, scope: HostScope = <factory>)¶

Bases: object

The active per-invocation runtime: the chosen lab, runtime flags, and the host lifecycle scope.

lab : Lab¶

dry_run : bool = False¶

log_command_output : bool = True¶

scope : HostScope¶

get_host(host_id: str, **overrides: Any) → UnixHost¶

all_hosts(pattern: re.Pattern[str] | None = None, *, include_containers: bool = False, **overrides: Any) → Iterator[RemoteHost]¶

async do_for_all_hosts(method: Callable[[...], Awaitable[T]], *args: Any, pattern: Pattern[str] | None = None, concurrent: bool = True, include_containers: bool = False, term: str | None = None, transfer: str | None = None, ssh_options: Any | None = None, telnet_options: Any | None = None, sftp_options: Any | None = None, scp_options: Any | None = None, ftp_options: Any | None = None, nc_options: Any | None = None, **kwargs: Any) → dict[str, T | BaseException]¶

async run_on_all_hosts(cmds: list[str] | str, pattern: re.Pattern[str] | None = None, concurrent: bool = True, timeout: float | None = None, *, include_containers: bool = False, term: str | None = None, transfer: str | None = None, ssh_options: Any = None, telnet_options: Any = None, sftp_options: Any = None, scp_options: Any = None, ftp_options: Any = None, nc_options: Any = None) → dict[str, RunResult | BaseException]¶

otto.context.open_context(*, lab: Lab | str | list[str], dry_run: bool = False, log_command_output: bool = True, search_paths: list[Path] | None = None) → AsyncIterator[OttoContext]¶

Build, install, and tear down an OttoContext for library / script use.

Pass a Lab, or a lab name / list of names to load via load_lab. On exit the host scope closes any still-connected hosts and the contextvar is reset. Does NOT run a reservation check — that is a CLI concern; a script that wants one calls otto.reservations.check_reservations explicitly.