utils¶

Shared utilities: status enums, CLI overlay sentinels, and async helpers.

otto.utils.split_on_commas(values: list[str] | str) → list[str]¶

Split a string or list of strings on commas into a flat list.

Parameters:¶

values – A single comma-separated string, or a list of such strings.

Returns:¶

A flat list of the individual values.

>>> split_on_commas("a,b,c")
['a', 'b', 'c']
>>> split_on_commas(["a,b", "c,d"])
['a', 'b', 'c', 'd']
>>> split_on_commas("single")
['single']

otto.utils.complete_comma_list(candidates: list[str], incomplete: str) → list[str]¶

Filter candidates for tab-completing one entry of a comma-separated option.

Options like --lab a,b and --tests x,y take a comma-joined value, which the shell hands to the completer as a single incomplete word. Complete only the final (in-progress) segment, keep the already-typed prefix intact, and drop candidates already present earlier in the list so completion never re-offers them.

>>> complete_comma_list(["tech1", "tech2", "prod"], "tech")
['tech1', 'tech2']
>>> complete_comma_list(["tech1", "tech2", "prod"], "tech1,te")
['tech1,tech2']

otto.utils.async_typer_command(f: Callable[[P], Coroutine[Any, Any, R]]) → Callable[[P], R]¶

Wrap an async Typer command so it runs under the active OttoContext scope.

Calls asyncio.run on the coroutine. If an OttoContext is already open (i.e. try_get_context() returns one), the coroutine runs inside its async context manager scope; otherwise it runs bare.

class otto.utils.Arg(variadic: bool = False, elem_type: type | None = None, name: str | None = None, help: str | None = None)¶

Bases: object

CLI overlay: force a parameter to a positional argument.

variadic=True makes it a space-separated list of elem_type (used for Python-union list params Typer can’t read, e.g. str | Sequence[...]). elem_type also overrides the CLI type for a scalar union. name sets the argument’s displayed metavar (e.g. Arg(name="SOURCE") shows SOURCE in --help instead of the parameter’s Python name); it does not change how the argument is passed, since positional arguments have no flag. Imports no typer.

variadic : bool = False¶

elem_type : type | None = None¶

name : str | None = None¶

help : str | None = None¶

class otto.utils.Opt(elem_type: type | None = None, name: str | None = None, help: str | None = None)¶

Bases: object

CLI overlay: force a parameter to a --option. Imports no typer.

name overrides the synthesized flag (e.g. Opt(name="--dest") makes the CLI flag --dest regardless of the parameter’s Python name); the value is still bound to the original Python parameter. When omitted, the flag defaults to --<param-name> as usual.

elem_type : type | None = None¶

name : str | None = None¶

help : str | None = None¶

otto.utils.cli_exposed(fn: Callable[[...], Any] | None = None, *, name: str | None = None, help_: str | None = None, success: str | None = None, output_dir: bool = True) → Callable[[...], Any]¶

Mark a host coroutine method for auto-exposure as an otto host subcommand.

name defaults to the method name with underscores dashed. success is an optional message printed on a successful (Status, "") result (e.g. “Transfer complete.”). output_dir=False marks a read-only verb that creates no per-invocation output directory (e.g. exists/lsmod); the default True keeps one.

Usable bare (@cli_exposed) or called (@cli_exposed(name=..., ...)).

otto.utils.is_literal(value: Any, literal_type: type[T]) → T¶

Raise a TypeError if value is not a valid member of the Literal type.

class otto.utils.Status(value)¶

Bases: Enum

General status enum for commands and tests.

>>> Status.Success
<Status.Success: 0>
>>> Status.Failed
<Status.Failed: 1>
>>> Status(0) is Status.Success
True

Success = 0¶

Failed = 1¶

Error = 2¶

Unstable = 3¶

Skipped = 4¶

property is_ok : bool¶

True for statuses that should be treated as passing (Success, Skipped).