host.app_shell¶

AppShell parsing engine — Parsed models and the parse= dispatch.

This module holds both halves of otto’s AppShell feature: the parsing half — regex-backed pydantic models (Parsed) and the functions that turn REPL output into typed objects — and the interactive AppShell REPL that wraps an application (mysql, python3) living inside an already-open shell session, driving it with AppShell.cmd() and locking out the session’s sentinel-framed run while attached.

A Parsed subclass pairs a pydantic model with the compiled regex that produces it. Named groups feed same-named fields; pydantic converts the captured strings to the field types. A field typed as another Parsed subclass — or list[Sub] / Sub | None — is parsed recursively from the region its group captured, so composite REPL output (mysql’s bordered table and its trailing stats line, say) maps to a nested object graph.

The public entry point is apply_parse(), which dispatches on the shape of the parse= spec:

a Parsed subclass -> single parse_one() (pattern.search);
list[Sub] -> parse_all() (pattern.finditer), where the empty list is a valid “zero rows” answer;
any other callable -> called as an escape hatch, its return value used verbatim and any exception surfaced as ParseMismatch.

exception otto.host.app_shell.ParseMismatch¶

Bases: ValueError

Output did not match the model’s pattern (or the callable raised).

class otto.host.app_shell.Parsed¶

Bases: OttoModel

A pydantic model plus the regex that produces it.

Named groups feed same-named fields; a field typed as another Parsed subclass (or list[Sub] / Sub | None) is recursively parsed from the region its group captured. Subclasses must define pattern as a compiled re.Pattern; a class-definition-time check enforces that the pattern’s named groups are a subset of the field names (typo guard) and a superset of the required fields (so pattern/model drift is impossible). Because every subclass self-checks at its own definition, nested models are validated automatically at each level.

pattern : ClassVar[Pattern[str]]¶

otto.host.app_shell.parse_one(model: type[Parsed], text: str) → Parsed¶

Search text with model.pattern and build a single instance.

Raises ParseMismatch (naming the pattern) if nothing matches.

otto.host.app_shell.parse_all(model: type[Parsed], text: str) → list[Parsed]¶: Return one model per finditer match; the empty list is valid.

otto.host.app_shell.apply_parse(spec: Any, text: str) → Any¶

Apply a parse= spec to text and return the parsed value.

spec is a Parsed subclass (single parse_one()), a list[Sub] of one (parse_all()), or any other callable used as an escape hatch — its exceptions are wrapped in ParseMismatch.

exception otto.host.app_shell.AppShellActiveError¶

Bases: RuntimeError

A shell session already has an AppShell attached.

Raised by run_cmd() while a shell is attached — the sentinel command frame must never be typed into the app, so run() is locked out — and by AppShell.attach() when the session already owns an active shell (AppShells do not nest).

exception otto.host.app_shell.AppShellTimeoutError¶

Bases: TimeoutError

The application REPL’s prompt did not return within the timeout.

A state failure (the REPL is left in an unknown state), distinct from a data mismatch: AppShell.cmd() marks the shell broken and the attaching context manager, on unwind, skips quit_cmd and goes straight to the command-frame recovery handshake.

On the caller-owned AppShell.attach() path that recovery cannot always confirm the POSIX shell is back; discard the session after this error rather than reusing it (see AppShell.attach()).

class otto.host.app_shell.AppShell(session: HostSession, timeout: float | None = None)¶

Bases: object

Base class for an application REPL living inside a shell session.

A subclass declares how to start the app (launch), how to recognise its prompt (prompt), and how to quit (quit_cmd); it may add methods that build on cmd(). Attach it to an already-open HostSession with the attach() async context manager, then drive the REPL with cmd():

class MySql(AppShell):
    launch = "mysql --pager=cat"
    prompt = re.compile(r"mysql> \Z")
    quit_cmd = "quit"


async with MySql.attach(session) as sql:
    rows = (await sql.cmd("SELECT id FROM users;", parse=list[Row])).value

While attached, the session’s sentinel-framed run() is locked out (raises AppShellActiveError) so the command frame is never typed into the app; raw send / expect stay available. AppShell is a plain class, not a pydantic model.

launch : ClassVar[str]¶: Command that starts the application REPL (sent with a trailing newline).

prompt : ClassVar[Pattern[str]]¶: Pattern matching the app’s prompt. A str is compiled at class-definition time; an end-anchored pattern (r"...\Z") is recommended so a prompt-looking substring in the output cannot match early.

quit_cmd : ClassVar[str] = 'exit'¶: Line sent on a clean context exit to leave the REPL.

user : ClassVar[str | None] = None¶: Cred login to become before launching (Part-1 proxy machinery); None keeps the session’s current user.

cmd_timeout : ClassVar[float] = 30.0¶: Class-level default seconds to wait for the prompt on launch and cmd(). Overridable per-session via timeout= on attach() / app_shell() (governs the launch wait and becomes the default for every cmd() in that session), and per-command via cmd(timeout=) (wins over both).

classmethod attach(session: HostSession, *, timeout: float | None = None) → AsyncIterator[Self]¶

Attach the shell to an already-open session (async context manager).

Takes the session’s app-shell lock, sends launch, waits for prompt, and yields the shell instance. While the block runs, the session’s sentinel-framed run is locked out. On exit it sends quit_cmd (unless the shell is broken), confirms the POSIX shell via the command-frame recovery handshake, and releases the lock; the session itself is left open. Raises AppShellActiveError if the session already has a shell attached, or AppShellTimeoutError if the launch prompt never arrives.

timeout, if given, becomes this session’s default prompt-wait: it governs the launch wait below and is used by cmd() for every call in the session that doesn’t pass its own timeout=. Falls back to cmd_timeout when omitted.

Note

After an AppShellTimeoutError, treat this session as spent: discard it and open a fresh one rather than reusing it. On this caller-owned attach path the shell may be left parked inside the application REPL, and the recovery handshake cannot always confirm the POSIX shell is back — a REPL that ignores Ctrl+C and echoes the recovery marker back can spoof a successful recovery. app_shell() is unaffected: it owns and closes the session for you.

async cmd(text: str, *, parse: Any | None = None, timeout: float | None = None) → ShellResult¶

Run one line in the REPL and return its ShellResult.

Sends text, waits for the next prompt, and strips the echoed command line (if the app echoed it), ANSI sequences, and the matched prompt — what remains is output. With no parse the output is also the value; with a parse spec the output is fed to apply_parse(). A parse mismatch is a data problem — it returns a failed ShellResult with the output preserved, not an exception. A prompt timeout is a state problem — the shell is marked broken and AppShellTimeoutError is raised.

timeout, if given, overrides this session’s default prompt-wait for this call only; otherwise the session default set at attach() time is used (which itself falls back to cmd_timeout).