API Reference

The names below are the public exports from hostctl.__all__ — everything a caller has to be able to write: a host or config to construct, an exception to catch, a type to annotate with, an argument to pass, or a contract to implement.

Concrete backends and result types the library only ever hands back are not listed here and are not part of the stable surface. They remain importable from the module that defines them — hostctl.host.qemu, hostctl.host.container_path, hostctl.host._winrm, hostctl.process, hostctl.executor, and hostctl.provider.transports — but their names and signatures may change without a deprecation cycle.

Hosts and configuration

hostctl.Host

Bases: ABC

Protocol-independent operational interface to a machine.

capabilities abstractmethod property

Operations supported by this host.

executor property

The command executor used when binding this host's shell.

executor_capabilities property

Native context/argument features of the underlying executor.

Derived from the host's own executor so a host that does not compose providers still reports truthfully. Host.executor's default wrapper reads this property, so it is skipped here to avoid recursing; a host that overrides executor with a real executor reports that executor's capabilities.

last_selection property

The redacted provider trace for the most recent :meth:run.

The run-side counterpart of CompositePosixPath.selection_trace. Entries appear in provider precedence order and carry provider, availability, reason, capabilities, chosen, generation, policy, and pin.

The trace accumulates across the failover attempts of a single run(), so a call that fell through from one provider to another reports every provider tried and every refusal reason -- including on the very call that suffered them.

Empty for a host whose run() does not select between providers (QemuHost, SerialHost), and empty before the first run(). Values are redacted on the way in; see :meth:~hostctl.provider.ProviderSelector.redact for the limits of that guarantee.

shell_flavour property

The explicitly known shell language used by this host.

close()

Close transport resources; must be safe to call repeatedly.

Source code in src/hostctl/host/_common.py

def close(self) -> None:
    """Close transport resources; must be safe to call repeatedly."""

connect()

Open transport resources; local/default implementations are no-op.

Source code in src/hostctl/host/_common.py

def connect(self) -> None:
    """Open transport resources; local/default implementations are no-op."""

info() abstractmethod

Return normalized system information without inferred values.

Source code in src/hostctl/host/_common.py

@_abc.abstractmethod
def info(self) -> HostInfo:
    """Return normalized system information without inferred values."""

path(*segments, backend=None)

Return a pathlib-compatible path for this host.

Source code in src/hostctl/host/_common.py

def path(self, *segments: PathLike, backend: _ty.Optional[str] = None) -> HostPath:
    """Return a pathlib-compatible path for this host."""
    raise NotImplementedError(
        f"{type(self).__name__} does not provide the 'path' capability"
    )

run(*cmds, bufsize=-1, executable=None, stdin=None, stdout=None, stderr=None, cwd=None, env=None, capture_output=True, check=True, encoding=None, errors=None, input=None, timeout=None, text=None)

Run commands and return a subprocess-compatible result.

A string is verbatim shell text. A tuple/list is one quoted argv command. A leading :class:pathlib.PurePath/pathlib_next path is a direct executable and all trailing values are its argv arguments. Otherwise multiple top-level commands are joined by the selected shell's command separator.

Source code in src/hostctl/host/_common.py

def run(
    self,
    *cmds: Command,
    bufsize: int = -1,
    executable: _ty.Optional[str] = None,
    stdin: _ty.Optional[FileHandle] = None,
    stdout: _ty.Optional[FileHandle] = None,
    stderr: _ty.Optional[FileHandle] = None,
    cwd: _ty.Optional[PathLike] = None,
    env: _ty.Optional[Environment] = None,
    capture_output: CaptureOutput = True,
    check: bool = True,
    encoding: _ty.Optional[str] = None,
    errors: _ty.Optional[str] = None,
    input: Input = None,
    timeout: _ty.Optional[float] = None,
    text: _ty.Optional[bool] = None,
) -> _subprocess.CompletedProcess:
    """Run commands and return a subprocess-compatible result.

    A string is verbatim shell text.  A tuple/list is one quoted argv
    command.  A leading :class:`pathlib.PurePath`/``pathlib_next`` path is
    a direct executable and all trailing values are its argv arguments.
    Otherwise multiple top-level commands are joined by the selected
    shell's command separator.
    """
    raise NotImplementedError(
        f"{type(self).__name__} does not provide the 'run' capability"
    )

shell()

A shell bound to this host's executor.

Used directly -- host.shell.run(...), host.shell.session(...), or with host.shell as session: -- it carries no defaults.

Called with keywords -- host.shell(cwd="/srv/app", env={"TZ": "UTC"}) -- it returns a shell carrying those defaults, applied to every later run, execute, and session that does not pass its own value. env merges per key; cwd, encoding, and errors override.

Source code in src/hostctl/host/_common.py

@_ShellAccessor
def shell(self) -> Shell[_subprocess.CompletedProcess]:
    """A shell bound to this host's executor.

    Used directly -- `host.shell.run(...)`, `host.shell.session(...)`, or
    `with host.shell as session:` -- it carries no defaults.

    Called with keywords -- `host.shell(cwd="/srv/app", env={"TZ": "UTC"})`
    -- it returns a shell carrying those defaults, applied to every later
    `run`, `execute`, and `session` that does not pass its own value.
    `env` merges per key; `cwd`, `encoding`, and `errors` override.
    """
    from ..shell import Shell

    return Shell(self.shell_flavour, self)

spawn(*cmds, executable=None, cwd=None, env=None, terminal=None, encoding=None, errors=None)

Start a persistent process controlled through a synchronous facade.

Source code in src/hostctl/host/_common.py

def spawn(
    self,
    *cmds: Command,
    executable: _ty.Optional[str] = None,
    cwd: _ty.Optional[PathLike] = None,
    env: _ty.Optional[Environment] = None,
    terminal: TerminalRequest = None,
    encoding: _ty.Optional[str] = None,
    errors: _ty.Optional[str] = None,
) -> Process:
    """Start a persistent process controlled through a synchronous facade."""
    raise NotImplementedError(
        f"{type(self).__name__} does not provide the 'spawn' capability"
    )

hostctl.HostConfig()

Bases: ABC

Secret-safe connection configuration and extensible URI dispatch.

Source code in src/hostctl/host/_common.py

def __init__(self) -> None:
    self._opened_host: _ty.Optional[Host] = None
    self._lifecycle_lock = _threading.Lock()

connection_uri abstractmethod property

Credential-safe canonical connection URI.

open()

Return a host context manager for this configuration.

Source code in src/hostctl/host/_common.py

def open(self) -> Host:
    """Return a host context manager for this configuration."""
    return self._create_host()

hostctl.HostInfo(hostname=None, os_family=None, os_name=None, os_version=None, architecture=None) dataclass

Normalized system information; unavailable values remain None.

hostctl.LocalConfig()

Bases: HostConfig

Source code in src/hostctl/host/_local.py

def __init__(self) -> None:
    super().__init__()

hostctl.LocalHost(config=None, *, executor_providers=(), path_providers=())

Bases: Host

A host whose commands and paths are local to this process.

The public surface is unchanged, but execution and filesystem access are assembled from ordered providers (:class:LocalExecutorProvider and :class:LocalPathProvider) rather than a hard-wired executor. A subclass may supply its own providers to reuse the local semantics over a different access mechanism.

Source code in src/hostctl/host/_local.py

def __init__(
    self,
    config: _ty.Optional[LocalConfig] = None,
    *,
    executor_providers: _ty.Iterable[object] = (),
    path_providers: _ty.Iterable[object] = (),
) -> None:
    self.config = config or LocalConfig()
    self._executor_provider_selector = ProviderSelector(
        tuple(executor_providers) or (LocalExecutorProvider(),)
    )
    self._path_provider_selector = ProviderSelector(
        tuple(path_providers) or (LocalPathProvider(),)
    )

executor_providers property

The ordered command providers backing :meth:run.

path_providers property

The ordered filesystem providers backing :meth:path.

hostctl.SshConfig(host, port=22, username='root', password=None, client_keys=None, executable=None, known_hosts=(), dialect=POSIX_SHELL, path_flavor=PosixPathname) dataclass

Bases: HostConfig

Explicit SSH transport, authentication, and target-shell settings.

hostctl.WinRMConfig(host, username, password=None, transport='ntlm', port=None, ssl=False, server_cert_validation='validate', message_encryption='auto', operation_timeout_sec=20, read_timeout_sec=30, provider='auto') dataclass

Bases: HostConfig

Connection and transport settings for the Windows provider.

hostctl.WinRMPath(*segments, backend=None)

Bases: WindowsPathname, Path

A Windows path whose I/O executes through a WinRM host.

Source code in src/hostctl/host/_winrm.py

def __init__(self, *segments, backend=None):
    # Python 3.14's pathlib.PurePath.__init__ no longer accepts kwargs.
    # Path state is initialized by __new__; backend is attached there.
    if not hasattr(self, "_raw_paths") and not hasattr(self, "_parts"):
        _StdPurePath.__init__(self, *segments)

symlink_to(target, target_is_directory=False)

Create this path as a symbolic link to target.

Windows only permits this from an elevated session or with Developer Mode enabled; otherwise the backend raises :class:PermissionError. target_is_directory is accepted for :meth:pathlib.Path.symlink_to signature parity and ignored -- New-Item -ItemType SymbolicLink infers the kind from the target.

Source code in src/hostctl/host/_winrm.py

def symlink_to(self, target, target_is_directory: bool = False):
    """Create this path as a symbolic link to ``target``.

    Windows only permits this from an elevated session or with
    Developer Mode enabled; otherwise the backend raises
    :class:`PermissionError`.  ``target_is_directory`` is accepted for
    :meth:`pathlib.Path.symlink_to` signature parity and ignored --
    ``New-Item -ItemType SymbolicLink`` infers the kind from the target.
    """
    self.backend.symlink(str(self), str(target))

hostctl.ContainerConfig(container, engine_url=None, user=None, workdir=None, executable=None, dialect='auto', path_flavor='auto', client_factory=None) dataclass

Bases: HostConfig

Docker Engine target and in-container execution defaults.

hostctl.ContainerHost(config, *, executor_providers=(), path_providers=())

Bases: Host

A running container reached through the Docker Engine API.

Commands and paths are assembled from ordered providers (:class:ContainerExecutorProvider and :class:ContainerArchivePathProvider) without changing the public API. The archive provider declares only the operations the Docker archive API can perform, so an unsupported mutation is rejected outright instead of falling through to another provider.

Source code in src/hostctl/host/container.py

def __init__(
    self,
    config: ContainerConfig,
    *,
    executor_providers: typing.Iterable[object] = (),
    path_providers: typing.Iterable[object] = (),
) -> None:
    self.config = config
    self._client: typing.Optional[object] = None
    self._container: typing.Optional[ContainerLike] = None
    self._attrs: typing.Optional[typing.Mapping[str, object]] = None
    self._executor = ContainerExecutor(
        lambda: self.container,
        user=config.user,
        workdir=config.workdir,
    )
    self._executor_provider_selector = ProviderSelector(
        tuple(executor_providers)
        or (
            ContainerExecutorProvider(
                self._executor,
                connect=lambda: self.container,
                close=self._close_client,
            ),
        )
    )
    self._path_provider_selector = ProviderSelector(
        tuple(path_providers) or (ContainerArchivePathProvider(self._archive_path),)
    )

executor_providers property

The ordered command providers backing :meth:run.

path_providers property

The ordered filesystem providers backing :meth:path.

spawn(*cmds, executable=None, cwd=None, env=None, terminal=None, encoding=None, errors=None)

Start a persistent Docker exec process with an optional TTY.

Source code in src/hostctl/host/container.py

def spawn(
    self,
    *cmds: Command,
    executable: typing.Optional[str] = None,
    cwd: typing.Optional[PathLike] = None,
    env: typing.Optional[Environment] = None,
    terminal: TerminalRequest = None,
    encoding: typing.Optional[str] = None,
    errors: typing.Optional[str] = None,
) -> Process:
    """Start a persistent Docker exec process with an optional TTY."""
    direct = starts_direct_command(cmds)
    if direct is not None:
        command, args = direct
        invocation = [command_text(command), *(command_text(v) for v in args)]
        environment = normalize_environment(env)
    elif cmds:
        script = self.shell_flavour.script(cmds, cwd=None, env=None)
        invocation = self.shell_flavour.invocation(
            script, executable=executable or self.config.executable
        )
        environment = normalize_environment(env)
    else:
        if cwd is not None or env is not None:
            raise ValueError("cwd and env require a command when spawning")
        selected = executable or self.config.executable
        invocation = (
            [selected]
            if selected
            else list(self.shell_flavour.invocation("", executable=None)[:1])
        )
        environment = None

    selected_terminal = terminal_options(terminal)
    tty = selected_terminal is not None
    api = typing.cast(typing.Any, self.client).api
    selected_workdir = str(cwd) if cwd is not None else self.config.workdir
    options: typing.Dict[str, object] = {
        "cmd": list(invocation),
        "stdin": True,
        "stdout": True,
        "stderr": True,
        "tty": tty,
    }
    if environment is not None:
        options["environment"] = environment
    if selected_workdir is not None:
        options["workdir"] = selected_workdir
    if self.config.user is not None:
        options["user"] = self.config.user
    try:
        created = api.exec_create(
            getattr(self.container, "id", self.config.container),
            **options,
        )
        exec_id = created["Id"] if isinstance(created, dict) else created
        stream = api.exec_start(exec_id, socket=True, tty=tty)
        if selected_terminal is not None:
            try:
                api.exec_resize(
                    exec_id,
                    height=selected_terminal.rows,
                    width=selected_terminal.columns,
                )
            except Exception:
                close = getattr(stream, "close", None)
                if close is not None:
                    close()
                raise
    except Exception as exc:
        normalized = normalize_container_error(exc)
        if normalized is exc:
            raise
        raise normalized from exc
    return ContainerProcess(
        api,
        str(exec_id),
        stream,
        tty=tty,
        command=list(invocation),
        encoding=encoding,
        errors=errors,
    )

hostctl.QemuConfig(domain, transport='libvirt', connection=None, socket_path=None, ssh=None, agent_timeout=10.0, dialect='auto', path_flavor='auto', transport_factory=None, serial_console=None) dataclass

Bases: HostConfig

Guest identity, QGA transport, and guest semantic selections.

hostctl.QemuHost(config)

Bases: Host

A guest OS reached through QEMU Guest Agent.

Source code in src/hostctl/host/qemu.py

def __init__(self, config: QemuConfig) -> None:
    self.config = config
    self._transport: typing.Optional[GuestAgentTransport] = None
    self._ssh_transport: typing.Optional[_SshTransport] = None
    self._commands: typing.Optional[typing.FrozenSet[str]] = None
    self._os_info: typing.Mapping[str, object] = {}
    self._hostname: typing.Optional[str] = None
    self._executor = QemuExecutor(lambda: self.transport)
    self._path_backend: typing.Optional[QgaPathBackend] = None
    self._path_provider: typing.Optional[object] = None

path_provider property

The QGA path provider, once :meth:path has assembled it.

open_serial()

Open the explicitly configured raw VM console.

Source code in src/hostctl/host/qemu.py

def open_serial(self) -> Process:
    """Open the explicitly configured raw VM console."""
    if self.config.serial_console is None:
        raise NotImplementedError("QEMU host has no configured serial console")
    return self.config.serial_console.open()

hostctl.SerialConfig(port, baudrate=115200, bytesize=8, parity='N', stopbits=1, xonxoff=False, rtscts=False, dsrdtr=False, read_timeout=0.1, write_timeout=10, inter_byte_timeout=None, exclusive=None, protocol=RawConsoleProfile(), username=None, password=None, serial_factory=None, serial_port=None) dataclass

Bases: HostConfig

Serial transport settings; credentials and profiles stay out of URIs.

hostctl.SerialHost(config)

Bases: Host

Host facade over one exclusive serial console byte stream.

Source code in src/hostctl/host/serial.py

def __init__(self, config: SerialConfig) -> None:
    self.config = config
    settings = SerialSettings(
        config.port,
        config.baudrate,
        config.bytesize,
        config.parity,
        config.stopbits,
        config.xonxoff,
        config.rtscts,
        config.dsrdtr,
        config.read_timeout,
        config.write_timeout,
        config.inter_byte_timeout,
        config.exclusive,
    )
    self._executor = SerialExecutor(
        settings,
        serial_factory=config.serial_factory,
        serial_port=config.serial_port,
        owns_serial_port=False if config.serial_port is not None else True,
    )
    self._negotiated = False

hostctl.SystemConfig(authority='localhost', *, shell=None, executor=(), path=(), **options)

Bases: HostConfig

Base configuration for a logical system with one or more providers.

Abstract: concrete system families are :class:PosixConfig, :class:WindowsConfig, and :class:IosConfig. Each binds host_type and uri_scheme; this class binds neither and cannot be instantiated into a host on its own.

Source code in src/hostctl/host/system.py

def __init__(
    self,
    authority: str = "localhost",
    *,
    shell: object = None,
    executor: typing.Iterable[str] = (),
    path: typing.Iterable[str] = (),
    **options: object,
):
    super().__init__()
    self.authority = authority or "localhost"
    self.shell = shell
    self.executors = tuple(executor)
    self.paths = tuple(path)
    self.options = dict(options)
    self._provider_transports = {}

hostctl.SystemHost(config=None, *, executor_providers=(), path_providers=(), shell=None, info=None, initializer=None)

Bases: Host

Host orchestration shared by POSIX, Windows, and IOS systems.

Source code in src/hostctl/host/system.py

def __init__(
    self,
    config: SystemConfig | None = None,
    *,
    executor_providers=(),
    path_providers=(),
    shell=None,
    info: HostInfo | None = None,
    initializer=None,
):
    # A config-less host builds the configuration matching its own system
    # family.  Previously this made a bare `SystemConfig` and then
    # assigned to `config.scheme` -- writing to what `HostConfig` defines
    # as a read-only computed property, which only worked because
    # `SystemConfig` shadowed that property with a plain string.  Picking
    # the right config type instead keeps `scheme` derived from the URI.
    self.config = config if config is not None else self.config_type()
    self._executor_selector = ProviderSelector(executor_providers)
    self._path_selector = ProviderSelector(path_providers)
    self._shell_resolver = (
        shell if callable(shell) and not isinstance(shell, type) else None
    )
    self._shell = (
        None
        if self._shell_resolver is not None
        else (
            shell_flavour(shell)
            if shell is not None
            else (
                shell_flavour(getattr(self.config, "shell", None))
                if getattr(self.config, "shell", None)
                else self.default_shell
            )
        )
    )
    self._info = info
    config_options = getattr(self.config, "options", {})
    self._initializer = (
        initializer
        if initializer is not None
        else (
            config_options.get("initializer")
            if isinstance(config_options, dict)
            else None
        )
    )
    if self._initializer is not None and not callable(self._initializer):
        raise TypeError("initializer must be callable")
    self._initializer_generation = False
    self._connected = False
    self._connected_providers = []
    self._closed_targets = set()
    # Serializes this host's connection *bookkeeping*: `_connected`,
    # `_connected_providers`, `_closed_targets`, and `_initializer_
    # generation`.  Without it, concurrent `run()` calls race the
    # check-then-append in `_ensure_provider_connected` and each append a
    # duplicate entry for the same provider, so the list grows without
    # bound and every racing caller repeats the connect round-trip.
    #
    # RLock, not Lock: these paths nest.  `connect()` runs the session
    # initializer, which is handed this same host and legitimately calls
    # `run()` -> `_ensure_provider_connected`; a plain Lock would deadlock
    # on that re-entry.
    #
    # Scope: the lock deliberately spans the provider `connect()` call.
    # Connecting is the operation being deduplicated, so releasing the
    # lock around it would reintroduce the very race it exists to close --
    # two callers would both observe "not connected" and both dial out.
    # Providers already own the slow part behind their own locks
    # (`_SshTransport._ssh_lock`), and a `SystemHost` is one logical
    # target whose providers are ordered fallbacks for that same target,
    # not independent endpoints to be dialed in parallel.  Command
    # dispatch itself -- `run()`, `path()`, `spawn()` -- stays outside the
    # lock, so this never serializes actual remote work.
    self._lifecycle_lock = threading.RLock()

provider_details property

Return deterministic, non-dispatching provider availability details.

runspace()

Return a provider-owned typed runspace when one is available.

Source code in src/hostctl/host/system.py

def runspace(self):
    """Return a provider-owned typed runspace when one is available."""
    selected = self._executor_selector.select(capability="runspace")
    provider = selected.provider
    method = getattr(provider, "runspace", None)
    if method is None:
        raise NotImplementedError(
            f"executor provider {provider.name!r} does not support runspaces"
        )
    return method()

hostctl.PosixConfig(authority='localhost', *, shell=None, executor=(), path=(), **options)

Bases: SystemConfig

Source code in src/hostctl/host/system.py

def __init__(
    self,
    authority: str = "localhost",
    *,
    shell: object = None,
    executor: typing.Iterable[str] = (),
    path: typing.Iterable[str] = (),
    **options: object,
):
    super().__init__()
    self.authority = authority or "localhost"
    self.shell = shell
    self.executors = tuple(executor)
    self.paths = tuple(path)
    self.options = dict(options)
    self._provider_transports = {}

hostctl.PosixHost(config=None, *, executor_providers=(), path_providers=(), shell=None, info=None, initializer=None)

Bases: SystemHost

Source code in src/hostctl/host/system.py

def __init__(
    self,
    config: SystemConfig | None = None,
    *,
    executor_providers=(),
    path_providers=(),
    shell=None,
    info: HostInfo | None = None,
    initializer=None,
):
    # A config-less host builds the configuration matching its own system
    # family.  Previously this made a bare `SystemConfig` and then
    # assigned to `config.scheme` -- writing to what `HostConfig` defines
    # as a read-only computed property, which only worked because
    # `SystemConfig` shadowed that property with a plain string.  Picking
    # the right config type instead keeps `scheme` derived from the URI.
    self.config = config if config is not None else self.config_type()
    self._executor_selector = ProviderSelector(executor_providers)
    self._path_selector = ProviderSelector(path_providers)
    self._shell_resolver = (
        shell if callable(shell) and not isinstance(shell, type) else None
    )
    self._shell = (
        None
        if self._shell_resolver is not None
        else (
            shell_flavour(shell)
            if shell is not None
            else (
                shell_flavour(getattr(self.config, "shell", None))
                if getattr(self.config, "shell", None)
                else self.default_shell
            )
        )
    )
    self._info = info
    config_options = getattr(self.config, "options", {})
    self._initializer = (
        initializer
        if initializer is not None
        else (
            config_options.get("initializer")
            if isinstance(config_options, dict)
            else None
        )
    )
    if self._initializer is not None and not callable(self._initializer):
        raise TypeError("initializer must be callable")
    self._initializer_generation = False
    self._connected = False
    self._connected_providers = []
    self._closed_targets = set()
    # Serializes this host's connection *bookkeeping*: `_connected`,
    # `_connected_providers`, `_closed_targets`, and `_initializer_
    # generation`.  Without it, concurrent `run()` calls race the
    # check-then-append in `_ensure_provider_connected` and each append a
    # duplicate entry for the same provider, so the list grows without
    # bound and every racing caller repeats the connect round-trip.
    #
    # RLock, not Lock: these paths nest.  `connect()` runs the session
    # initializer, which is handed this same host and legitimately calls
    # `run()` -> `_ensure_provider_connected`; a plain Lock would deadlock
    # on that re-entry.
    #
    # Scope: the lock deliberately spans the provider `connect()` call.
    # Connecting is the operation being deduplicated, so releasing the
    # lock around it would reintroduce the very race it exists to close --
    # two callers would both observe "not connected" and both dial out.
    # Providers already own the slow part behind their own locks
    # (`_SshTransport._ssh_lock`), and a `SystemHost` is one logical
    # target whose providers are ordered fallbacks for that same target,
    # not independent endpoints to be dialed in parallel.  Command
    # dispatch itself -- `run()`, `path()`, `spawn()` -- stays outside the
    # lock, so this never serializes actual remote work.
    self._lifecycle_lock = threading.RLock()

from_ssh(config) classmethod

Compose POSIX semantics over an existing :class:SshConfig.

Source code in src/hostctl/host/system.py

@classmethod
def from_ssh(cls, config):
    """Compose POSIX semantics over an existing :class:`SshConfig`."""
    host = config._create_host()
    if not isinstance(host, cls):
        raise ValueError(
            f"SSH configuration selects {type(host).__name__}, not {cls.__name__}"
        )
    return host

hostctl.WindowsConfig(authority='localhost', *, shell=None, executor=(), path=(), **options)

Bases: SystemConfig

Source code in src/hostctl/host/system.py

def __init__(
    self,
    authority: str = "localhost",
    *,
    shell: object = None,
    executor: typing.Iterable[str] = (),
    path: typing.Iterable[str] = (),
    **options: object,
):
    super().__init__()
    self.authority = authority or "localhost"
    self.shell = shell
    self.executors = tuple(executor)
    self.paths = tuple(path)
    self.options = dict(options)
    self._provider_transports = {}

hostctl.WindowsHost(config=None, *, executor_providers=(), path_providers=(), shell=None, info=None, initializer=None)

Bases: SystemHost

Source code in src/hostctl/host/system.py

def __init__(
    self,
    config: SystemConfig | None = None,
    *,
    executor_providers=(),
    path_providers=(),
    shell=None,
    info: HostInfo | None = None,
    initializer=None,
):
    # A config-less host builds the configuration matching its own system
    # family.  Previously this made a bare `SystemConfig` and then
    # assigned to `config.scheme` -- writing to what `HostConfig` defines
    # as a read-only computed property, which only worked because
    # `SystemConfig` shadowed that property with a plain string.  Picking
    # the right config type instead keeps `scheme` derived from the URI.
    self.config = config if config is not None else self.config_type()
    self._executor_selector = ProviderSelector(executor_providers)
    self._path_selector = ProviderSelector(path_providers)
    self._shell_resolver = (
        shell if callable(shell) and not isinstance(shell, type) else None
    )
    self._shell = (
        None
        if self._shell_resolver is not None
        else (
            shell_flavour(shell)
            if shell is not None
            else (
                shell_flavour(getattr(self.config, "shell", None))
                if getattr(self.config, "shell", None)
                else self.default_shell
            )
        )
    )
    self._info = info
    config_options = getattr(self.config, "options", {})
    self._initializer = (
        initializer
        if initializer is not None
        else (
            config_options.get("initializer")
            if isinstance(config_options, dict)
            else None
        )
    )
    if self._initializer is not None and not callable(self._initializer):
        raise TypeError("initializer must be callable")
    self._initializer_generation = False
    self._connected = False
    self._connected_providers = []
    self._closed_targets = set()
    # Serializes this host's connection *bookkeeping*: `_connected`,
    # `_connected_providers`, `_closed_targets`, and `_initializer_
    # generation`.  Without it, concurrent `run()` calls race the
    # check-then-append in `_ensure_provider_connected` and each append a
    # duplicate entry for the same provider, so the list grows without
    # bound and every racing caller repeats the connect round-trip.
    #
    # RLock, not Lock: these paths nest.  `connect()` runs the session
    # initializer, which is handed this same host and legitimately calls
    # `run()` -> `_ensure_provider_connected`; a plain Lock would deadlock
    # on that re-entry.
    #
    # Scope: the lock deliberately spans the provider `connect()` call.
    # Connecting is the operation being deduplicated, so releasing the
    # lock around it would reintroduce the very race it exists to close --
    # two callers would both observe "not connected" and both dial out.
    # Providers already own the slow part behind their own locks
    # (`_SshTransport._ssh_lock`), and a `SystemHost` is one logical
    # target whose providers are ordered fallbacks for that same target,
    # not independent endpoints to be dialed in parallel.  Command
    # dispatch itself -- `run()`, `path()`, `spawn()` -- stays outside the
    # lock, so this never serializes actual remote work.
    self._lifecycle_lock = threading.RLock()

from_winrm(config) classmethod

Compose Windows semantics over an existing :class:WinRMConfig.

Source code in src/hostctl/host/system.py

@classmethod
def from_winrm(cls, config):
    """Compose Windows semantics over an existing :class:`WinRMConfig`."""
    host = config._create_host()
    if not isinstance(host, cls):
        raise ValueError(
            f"WinRM configuration selects {type(host).__name__}, not {cls.__name__}"
        )
    return host

hostctl.IosConfig(authority='localhost', *, shell=None, executor=(), path=(), **options)

Bases: SystemConfig

Source code in src/hostctl/host/system.py

def __init__(
    self,
    authority: str = "localhost",
    *,
    shell: object = None,
    executor: typing.Iterable[str] = (),
    path: typing.Iterable[str] = (),
    **options: object,
):
    super().__init__()
    self.authority = authority or "localhost"
    self.shell = shell
    self.executors = tuple(executor)
    self.paths = tuple(path)
    self.options = dict(options)
    self._provider_transports = {}

hostctl.IosHost(config=None, *, executor_providers=(), path_providers=(), shell=None, info=None, initializer=None)

Bases: SystemHost

Source code in src/hostctl/host/system.py

def __init__(
    self,
    config: SystemConfig | None = None,
    *,
    executor_providers=(),
    path_providers=(),
    shell=None,
    info: HostInfo | None = None,
    initializer=None,
):
    # A config-less host builds the configuration matching its own system
    # family.  Previously this made a bare `SystemConfig` and then
    # assigned to `config.scheme` -- writing to what `HostConfig` defines
    # as a read-only computed property, which only worked because
    # `SystemConfig` shadowed that property with a plain string.  Picking
    # the right config type instead keeps `scheme` derived from the URI.
    self.config = config if config is not None else self.config_type()
    self._executor_selector = ProviderSelector(executor_providers)
    self._path_selector = ProviderSelector(path_providers)
    self._shell_resolver = (
        shell if callable(shell) and not isinstance(shell, type) else None
    )
    self._shell = (
        None
        if self._shell_resolver is not None
        else (
            shell_flavour(shell)
            if shell is not None
            else (
                shell_flavour(getattr(self.config, "shell", None))
                if getattr(self.config, "shell", None)
                else self.default_shell
            )
        )
    )
    self._info = info
    config_options = getattr(self.config, "options", {})
    self._initializer = (
        initializer
        if initializer is not None
        else (
            config_options.get("initializer")
            if isinstance(config_options, dict)
            else None
        )
    )
    if self._initializer is not None and not callable(self._initializer):
        raise TypeError("initializer must be callable")
    self._initializer_generation = False
    self._connected = False
    self._connected_providers = []
    self._closed_targets = set()
    # Serializes this host's connection *bookkeeping*: `_connected`,
    # `_connected_providers`, `_closed_targets`, and `_initializer_
    # generation`.  Without it, concurrent `run()` calls race the
    # check-then-append in `_ensure_provider_connected` and each append a
    # duplicate entry for the same provider, so the list grows without
    # bound and every racing caller repeats the connect round-trip.
    #
    # RLock, not Lock: these paths nest.  `connect()` runs the session
    # initializer, which is handed this same host and legitimately calls
    # `run()` -> `_ensure_provider_connected`; a plain Lock would deadlock
    # on that re-entry.
    #
    # Scope: the lock deliberately spans the provider `connect()` call.
    # Connecting is the operation being deduplicated, so releasing the
    # lock around it would reintroduce the very race it exists to close --
    # two callers would both observe "not connected" and both dial out.
    # Providers already own the slow part behind their own locks
    # (`_SshTransport._ssh_lock`), and a `SystemHost` is one logical
    # target whose providers are ordered fallbacks for that same target,
    # not independent endpoints to be dialed in parallel.  Command
    # dispatch itself -- `run()`, `path()`, `spawn()` -- stays outside the
    # lock, so this never serializes actual remote work.
    self._lifecycle_lock = threading.RLock()

Executors and processes

hostctl.Exec(program, *args) dataclass

One command executed directly, with no shell layer.

Exec(program, *args) names a program and its argv. The program may be an absolute path or a bare name the target resolves through PATH; a bare name has no other spelling, since a plain string is always shell text.

This is the explicit marker for direct execution. A path used anywhere else is an ordinary value that stringifies, so several commands can be written without one of them silently becoming an executable::

host.run(Exec("/bin/ls", "-l"))       # one direct command
host.run(Exec("ls", "-l"))            # PATH-resolved, no shell
host.run(Exec("/bin/a"), Exec("/bin/b"))   # two direct commands
host.run(PurePosixPath("/bin/a"), PurePosixPath("/bin/b"))
                                      # two ordinary shell commands

Arguments are argv values, never nested commands: a list or tuple raises TypeError rather than blurring argv and shell semantics.

A deliberately non-iterable container. ShellFlavour.command_text dispatches structured commands on Iterable, so an iterable marker would be quoted into an argv string instead of taking the direct branch.

Source code in src/hostctl/host/_common.py

def __init__(self, program: ExecValue, *args: ExecValue) -> None:
    if not isinstance(program, (str, bytes, _PurePath, _Pathname)):
        raise TypeError("Exec program must be a str, bytes, or path value")
    for value in args:
        if isinstance(value, (tuple, list)):
            raise TypeError("direct command arguments must be scalar values")
        if not isinstance(value, (str, bytes, _PurePath, _Pathname)):
            raise TypeError(
                "direct command arguments must be str, bytes, or path values"
            )
    object.__setattr__(self, "program", program)
    object.__setattr__(self, "args", tuple(args))

hostctl.Executor

Bases: Protocol[_Result]

Callable executor receiving one command string and execution options.

hostctl.ExecutionOptions

Bases: TypedDict

Shell-agnostic subprocess-style options understood by executors.

hostctl.LocalExecutor

Bases: Executor[CompletedProcess]

Execute direct argv or finalized shell invocations locally.

hostctl.Process

Bases: Protocol

Synchronous control surface for a persistent child process.

hostctl.TerminalOptions(term_type='xterm-256color', columns=80, rows=24, pixel_width=0, pixel_height=0) dataclass

Requested pseudo-terminal type and initial dimensions.

hostctl.QgaCommandError(error_class, description, *, data=None)

Bases: QgaError

A structured QGA command error returned by the guest.

Source code in src/hostctl/executor/_qga.py

def __init__(
    self,
    error_class: str,
    description: str,
    *,
    data: typing.Optional[typing.Mapping[str, object]] = None,
) -> None:
    super().__init__(f"{error_class}: {description}")
    self.error_class = error_class
    self.description = description
    self.data = dict(data or {})

hostctl.QgaProtocolError

Bases: QgaError, ConnectionError

The guest agent returned malformed or inconsistent protocol data.

Shells and sessions

hostctl.ShellFlavour

Bases: ABC

Construct scripts and commands for one explicitly selected target shell.

change_directory(cwd) abstractmethod

Render a directory change which fails if the directory is absent.

Source code in src/hostctl/shell/_common.py

@abc.abstractmethod
def change_directory(self, cwd: PathLike) -> str:
    """Render a directory change which fails if the directory is absent."""

command(cmds, *, executable=None, cwd=None, env=None) abstractmethod

Build the command submitted to an SSH exec channel.

Source code in src/hostctl/shell/_common.py

@abc.abstractmethod
def command(
    self,
    cmds: typing.Iterable[ShellToken],
    *,
    executable: typing.Optional[str] = None,
    cwd: typing.Optional[PathLike] = None,
    env: typing.Optional[Environment] = None,
) -> ShellCommand:
    """Build the command submitted to an SSH exec channel."""

command_path(value)

Return a direct-command marker using the target shell's path syntax.

Source code in src/hostctl/shell/_common.py

def command_path(self, value: PathLike) -> PurePath:
    """Return a direct-command marker using the target shell's path syntax."""
    return self.path_flavor(value)

environment_assignment(key, value) abstractmethod

Render one validated environment assignment.

Source code in src/hostctl/shell/_common.py

@abc.abstractmethod
def environment_assignment(self, key: str, value: object) -> str:
    """Render one validated environment assignment."""

environment_script(env)

Convert an environment mapping into a standalone shell script.

Source code in src/hostctl/shell/_common.py

def environment_script(self, env: Environment) -> str:
    """Convert an environment mapping into a standalone shell script."""
    assignments = []
    for key, value in env.items():
        if isinstance(key, bytes):
            key = key.decode("utf-8", "surrogateescape")
        if not isinstance(key, str) or not _ENVIRONMENT_KEY.fullmatch(key):
            raise ValueError(f"invalid environment variable name: {key!r}")
        assignments.append(self.environment_assignment(key, value))
    return self.command_separator.join(assignments)

invocation(script, *, executable=None) abstractmethod

Build local-process argv which invokes this shell for one script.

Source code in src/hostctl/shell/_common.py

@abc.abstractmethod
def invocation(
    self,
    script: str,
    *,
    executable: typing.Optional[str] = None,
) -> typing.Sequence[str]:
    """Build local-process argv which invokes this shell for one script."""

join(values)

Join commands, preserving raw strings and explicit operators.

Source code in src/hostctl/shell/_common.py

def join(self, values: typing.Iterable[ShellToken]) -> str:
    """Join commands, preserving raw strings and explicit operators."""
    result = []
    pending = self.command_separator
    has_command = False
    expecting_command = False
    for value in values:
        if isinstance(value, ShellOperator):
            if not has_command or expecting_command:
                raise ValueError("shell operator must appear between commands")
            pending = self.operator(value)
            expecting_command = True
            continue
        command = self.command_text(value)
        if not command:
            continue
        if has_command:
            result.append(pending)
        result.append(command)
        has_command = True
        expecting_command = False
        pending = self.command_separator
    if expecting_command:
        raise ValueError("shell operator must be followed by a command")
    return "".join(result)

join_cwd(changed, command)

Join cwd setup and payload with the shell's AND operator.

Source code in src/hostctl/shell/_common.py

def join_cwd(self, changed: str, command: str) -> str:
    """Join cwd setup and payload with the shell's AND operator."""
    return f"{changed}{self.operator(ShellOperator.AND)}{command}"

operator(value) abstractmethod

Render one supported command operator.

Source code in src/hostctl/shell/_common.py

@abc.abstractmethod
def operator(self, value: ShellOperator) -> str:
    """Render one supported command operator."""

quote(value) abstractmethod

Quote one structured argument for this shell.

Source code in src/hostctl/shell/_common.py

@abc.abstractmethod
def quote(self, value: object) -> str:
    """Quote one structured argument for this shell."""

script(cmds, *, cwd=None, env=None, for_session=False)

Build a script, applying environment and cwd consistently.

Source code in src/hostctl/shell/_common.py

def script(
    self,
    cmds: typing.Iterable[ShellToken],
    *,
    cwd: typing.Optional[PathLike] = None,
    env: typing.Optional[Environment] = None,
    for_session: bool = False,
) -> str:
    """Build a script, applying environment and cwd consistently."""
    command = self.join(cmds)
    changed = self.change_directory(cwd) if cwd is not None else ""
    rendered = {
        "env": self.environment_script(env) if env else "",
        "cwd": changed,
        "command": command,
    }
    if (
        "cwd" in self.context_order
        and "command" in self.context_order
        and cwd is not None
        and command
    ):
        cwd_index = self.context_order.index("cwd")
        command_index = self.context_order.index("command")
        if command_index == cwd_index + 1:
            rendered["cwd"] = self.join_cwd(changed, command)
            rendered["command"] = ""
    parts = [rendered[name] for name in self.context_order if rendered[name]]
    script = self.command_separator.join(parts)
    epilogue = getattr(self, "execution_epilogue", "")
    if script and epilogue and not for_session:
        script += epilogue
    return script

hostctl.Shell(flavour, executor, *, cwd=None, env=None, encoding=None, errors=None)

Bases: Executor[_Result], Generic[_Result]

Bind one shell language to a one-string callable or host executor.

Source code in src/hostctl/shell/_common.py

def __init__(
    self,
    flavour: ShellFlavour,
    executor: typing.Union[Executor[_Result], Host],
    *,
    cwd: typing.Optional[PathLike] = None,
    env: typing.Optional[Environment] = None,
    encoding: typing.Optional[str] = None,
    errors: typing.Optional[str] = None,
) -> None:
    self.flavour = flavour
    #: Defaults applied to every `run`, `execute`, and `session` call that
    #: does not pass its own value.  `cwd`, `encoding`, and `errors`
    #: override wholesale; `env` merges per key so a call can change one
    #: variable without restating the rest (see `_resolve_env`).
    self.cwd = cwd
    self.env = dict(env) if env is not None else None
    self.encoding = encoding
    self.errors = errors
    run = getattr(executor, "run", None)
    spawn = getattr(executor, "spawn", None)
    self._spawn = spawn if callable(spawn) else None
    self._session: typing.Optional[ShellSession] = None
    if callable(executor):
        self._execute = executor
    elif callable(run):
        self._execute = run
    else:
        raise TypeError("executor must be callable or provide run(command)")
    try:
        self._executor_parameters = inspect.signature(self._execute).parameters
    except (TypeError, ValueError):
        self._executor_parameters = {}
    self._executor_accepts_options = any(
        item.kind is inspect.Parameter.VAR_KEYWORD
        for item in self._executor_parameters.values()
    )
    # A host `run(*cmds)` takes several *commands*; an executor
    # `__call__(command, *args)` takes one command and its argv. Only the
    # former needs the program and argv bundled into one `Exec`, since
    # otherwise each argument would become a separate command. The shapes
    # differ in their first parameter: VAR_POSITIONAL vs a named one.
    positional = [
        item
        for item in self._executor_parameters.values()
        if item.kind
        in (
            inspect.Parameter.POSITIONAL_ONLY,
            inspect.Parameter.POSITIONAL_OR_KEYWORD,
            inspect.Parameter.VAR_POSITIONAL,
        )
    ]
    self._execute_takes_commands = bool(positional) and (
        positional[0].kind is inspect.Parameter.VAR_POSITIONAL
    )
    published = getattr(executor, "executor_capabilities", None)
    if published is None:
        inferred = set()
        if self._accepts_keyword("cwd"):
            inferred.add(ExecutorCapability.CWD)
        if self._accepts_keyword("env"):
            inferred.add(ExecutorCapability.ENV)
        if any(
            item.kind is inspect.Parameter.VAR_POSITIONAL
            for item in self._executor_parameters.values()
        ):
            inferred.add(ExecutorCapability.ARGS)
        published = frozenset(inferred)
    # One capability vocabulary, strings -- see `ExecutorCapability`.
    # Its members subclass `str`, so a set published as enum members by a
    # raw `Executor` and one published as plain strings by a `Host` (via
    # `ExecutorProvider`) compare and hash identically.  No conversion
    # happens here, and none is needed at any other boundary.
    self.executor_capabilities = frozenset(published)
    self._executor_accepts_cwd = (
        ExecutorCapability.CWD in self.executor_capabilities
    )
    self._executor_accepts_env = (
        ExecutorCapability.ENV in self.executor_capabilities
    )

__enter__()

Open a default session, so with host.shell as session: works.

The session is closed on exit. session(...) remains the way to pass a command, cwd, env, terminal, or encoding; this is the no-argument shorthand. A Shell is not reusable as a context manager while a session it opened is still active -- each with opens its own.

Source code in src/hostctl/shell/_common.py

def __enter__(self) -> ShellSession:
    """Open a default session, so ``with host.shell as session:`` works.

    The session is closed on exit. `session(...)` remains the way to pass
    a command, cwd, env, terminal, or encoding; this is the no-argument
    shorthand. A `Shell` is not reusable as a context manager while a
    session it opened is still active -- each `with` opens its own.
    """
    if self._session is not None:
        raise RuntimeError("shell already has an active session")
    session = self.session()
    try:
        # Enter the session so the underlying process sees a balanced
        # __enter__/__exit__ pair; `session.__exit__` delegates to it.
        session.__enter__()
    except BaseException:
        session.close()
        raise
    self._session = session
    return session

configure(*, cwd=None, env=_INHERIT_ENV, encoding=None, errors=None)

Return a copy of this shell with additional defaults applied.

env merges over this shell's default the same way a per-call env does, so configuring twice layers rather than replaces, and None produces a copy carrying no environment default at all. The original shell is left unchanged, which keeps host.shell -- a fresh object per access -- safe to configure without surprising another caller.

Source code in src/hostctl/shell/_common.py

def configure(
    self,
    *,
    cwd: typing.Optional[PathLike] = None,
    env: EnvironmentSelection = _INHERIT_ENV,
    encoding: typing.Optional[str] = None,
    errors: typing.Optional[str] = None,
) -> "Shell[_Result]":
    """Return a copy of this shell with additional defaults applied.

    `env` merges over this shell's default the same way a per-call `env`
    does, so configuring twice layers rather than replaces, and `None`
    produces a copy carrying no environment default at all. The original
    shell is left unchanged, which keeps `host.shell` -- a fresh object
    per access -- safe to configure without surprising another caller.
    """
    clone = copy.copy(self)
    clone._session = None
    clone.cwd = self.cwd if cwd is None else cwd
    clone.env = self._resolve_env(env)
    clone.encoding = self.encoding if encoding is None else encoding
    clone.errors = self.errors if errors is None else errors
    return clone

execute(command, *args, stdin=None, stdout=None, stderr=None, cwd=None, env=None, capture_output=None, check=None, encoding=None, errors=None, input=None, timeout=None, text=None)

Execute one command and forward supported execution context.

Source code in src/hostctl/shell/_common.py

def execute(
    self,
    command: ExecutorCommand,
    *args: CommandArgument,
    stdin: typing.Optional[FileHandle] = None,
    stdout: typing.Optional[FileHandle] = None,
    stderr: typing.Optional[FileHandle] = None,
    cwd: typing.Optional[PathLike] = None,
    env: typing.Optional[Environment] = None,
    capture_output: typing.Optional[CaptureOutput] = None,
    check: typing.Optional[bool] = None,
    encoding: typing.Optional[str] = None,
    errors: typing.Optional[str] = None,
    input: Input = None,
    timeout: typing.Optional[float] = None,
    text: typing.Optional[bool] = None,
) -> _Result:
    """Execute one command and forward supported execution context."""
    executor_args = args
    if args and ExecutorCapability.ARGS not in self.executor_capabilities:
        command = self.flavour.structured_command((command, *args))
        executor_args = ()
    # Apply the shell's cwd/env defaults only where the executor can carry
    # them natively. Where it cannot, the caller renders them into the
    # script instead -- `run` does exactly that and then passes
    # cwd=None/env=None here, so resolving again would forward a value this
    # executor rejects. `execute` used directly against a capability-less
    # executor therefore ignores the defaults by design: it dispatches one
    # opaque command rather than building a script.
    if self._executor_accepts_cwd:
        cwd = self._resolve_cwd(cwd)
    if self._executor_accepts_env:
        env = self._resolve_env(env)
    if encoding is None:
        encoding = self.encoding
    if errors is None:
        errors = self.errors
    options = {
        name: value
        for name, value in (
            ("stdin", stdin),
            ("stdout", stdout),
            ("stderr", stderr),
            ("capture_output", capture_output),
            ("check", check),
            ("encoding", encoding),
            ("errors", errors),
            ("input", input),
            ("timeout", timeout),
            ("text", text),
        )
        if value is not None
    }
    if cwd is not None:
        if not self._executor_accepts_cwd:
            raise TypeError("executor does not accept cwd")
        options["cwd"] = cwd
    if env is not None:
        if not self._executor_accepts_env:
            raise TypeError("executor does not accept env")
        options["env"] = env
    unsupported = [
        name
        for name in options
        if not self._accepts_keyword(name) and not self._executor_accepts_options
    ]
    if unsupported:
        raise TypeError(f"executor does not accept {sorted(unsupported)[0]}")
    if executor_args and self._execute_takes_commands:
        # A host `run(*cmds)` treats several positionals as several
        # commands, so a program and its argv must arrive as one `Exec`
        # rather than relying on position to mark direct execution.
        # Only when there *are* argv values: `Shell.run` renders every
        # command into one script and dispatches it here with no args, and
        # that script is shell text to interpret, not a program to exec.
        return self._execute(Exec(command, *executor_args), **options)
    return self._execute(command, *executor_args, **options)

run(*cmds, stdin=None, stdout=None, stderr=None, cwd=None, env=_INHERIT_ENV, capture_output=None, check=None, encoding=None, errors=None, input=None, timeout=None, text=None)

Build one script from all commands and pass it to the executor.

env merges over the shell's default per key; the default merges nothing and so inherits it. Pass None to run without the shell's configured environment, keeping whatever the host provides itself.

Source code in src/hostctl/shell/_common.py

def run(
    self,
    *cmds: ShellToken,
    stdin: typing.Optional[FileHandle] = None,
    stdout: typing.Optional[FileHandle] = None,
    stderr: typing.Optional[FileHandle] = None,
    cwd: typing.Optional[PathLike] = None,
    env: EnvironmentSelection = _INHERIT_ENV,
    capture_output: typing.Optional[CaptureOutput] = None,
    check: typing.Optional[bool] = None,
    encoding: typing.Optional[str] = None,
    errors: typing.Optional[str] = None,
    input: Input = None,
    timeout: typing.Optional[float] = None,
    text: typing.Optional[bool] = None,
) -> _Result:
    """Build one script from all commands and pass it to the executor.

    `env` merges over the shell's default per key; the default merges
    nothing and so inherits it. Pass `None` to run without the shell's
    configured environment, keeping whatever the host provides itself.
    """
    # Resolve defaults here rather than in `execute`: the script is
    # rendered before dispatch, so an embedded `cd`/env assignment has to
    # see the shell's defaults too.
    cwd = self._resolve_cwd(cwd)
    env = self._resolve_env(env)
    script = self.flavour.script(
        cmds,
        cwd=None if self._executor_accepts_cwd else cwd,
        env=None if self._executor_accepts_env else env,
    )
    return self.execute(
        script,
        cwd=cwd if self._executor_accepts_cwd else None,
        env=env if self._executor_accepts_env else None,
        stdin=stdin,
        stdout=stdout,
        stderr=stderr,
        capture_output=capture_output,
        check=check,
        encoding=encoding,
        errors=errors,
        input=input,
        timeout=timeout,
        text=text,
    )

session(*cmds, executable=None, cwd=None, env=_INHERIT_ENV, terminal=None, encoding=None, errors=None)

Open a persistent process using this shell language.

env merges over the shell's default per key; the default merges nothing and so inherits it. Pass None to open the session without the shell's configured environment, keeping whatever the host provides itself.

Source code in src/hostctl/shell/_common.py

def session(
    self,
    *cmds: ShellToken,
    executable: typing.Optional[str] = None,
    cwd: typing.Optional[PathLike] = None,
    env: EnvironmentSelection = _INHERIT_ENV,
    terminal: TerminalRequest = None,
    encoding: typing.Optional[str] = None,
    errors: typing.Optional[str] = None,
) -> ShellSession:
    """Open a persistent process using this shell language.

    `env` merges over the shell's default per key; the default merges
    nothing and so inherits it. Pass `None` to open the session without
    the shell's configured environment, keeping whatever the host
    provides itself.
    """
    if self._spawn is None:
        raise NotImplementedError("executor does not provide persistent sessions")
    cwd = self._resolve_cwd(cwd)
    env = self._resolve_env(env)
    session = ShellSession(
        self.flavour,
        self._spawn(
            executable=executable,
            terminal=terminal,
            encoding=self.encoding if encoding is None else encoding,
            errors=self.errors if errors is None else errors,
        ),
    )
    # A shell default cwd/env applies to the session too: it is submitted
    # once here so it persists for every later `send` in that shell.
    if cmds or cwd is not None or env:
        session.send(
            *cmds,
            cwd=cwd,
            env=env,
        )
    return session

hostctl.ShellSession(flavour, process)

Bases: Process

A persistent process with shell-aware command submission.

Source code in src/hostctl/shell/_common.py

def __init__(self, flavour: ShellFlavour, process: Process) -> None:
    self.flavour = flavour
    self.process = process

send(*cmds, cwd=None, env=None)

Render and submit commands using this session's shell language.

Source code in src/hostctl/shell/_common.py

def send(
    self,
    *cmds: ShellToken,
    cwd: typing.Optional[PathLike] = None,
    env: typing.Optional[Environment] = None,
) -> None:
    """Render and submit commands using this session's shell language."""
    if not cmds and cwd is None and not env:
        raise ValueError("send requires commands, cwd, or env")
    script = self.flavour.script(cmds, cwd=cwd, env=env, for_session=True)
    self.process.write(
        script + self.flavour.command_separator + self.flavour.line_terminator
    )

hostctl.ShellOperator

Bases: Enum

hostctl.register_shell_flavour(value, *, name=None, replace=False)

Register a configured flavour instance for string-based selection.

Source code in src/hostctl/shell/__init__.py

def register_shell_flavour(
    value: typing.Union[ShellFlavour, typing.Type[ShellFlavour]],
    *,
    name: typing.Optional[str] = None,
    replace: bool = False,
) -> ShellFlavour:
    """Register a configured flavour instance for string-based selection."""
    flavour = value() if isinstance(value, type) else value
    if not isinstance(flavour, ShellFlavour):
        raise TypeError("shell flavour must be a ShellFlavour instance or subclass")
    selected_name = name or flavour.name
    if not selected_name:
        raise ValueError("shell flavour name must not be empty")
    if selected_name in _SHELLS and not replace:
        raise ValueError(f"shell flavour is already registered: {selected_name}")
    _SHELLS[selected_name] = flavour
    return flavour

hostctl.shell_flavour(value)

Normalize a public shell selection without inferring from the transport.

Source code in src/hostctl/shell/__init__.py

def shell_flavour(value: ShellFlavourSelection) -> ShellFlavour:
    """Normalize a public shell selection without inferring from the transport."""
    if isinstance(value, type):
        if not issubclass(value, ShellFlavour):
            raise TypeError("shell flavour class must inherit ShellFlavour")
        value = value()
    if isinstance(value, ShellFlavour):
        return value
    try:
        return _SHELLS[value]
    except (KeyError, TypeError) as exc:
        if isinstance(value, str):
            try:
                entries = _metadata.entry_points()
                selected = (
                    entries.select(group="hostctl.shell_flavours", name=value)
                    if hasattr(entries, "select")
                    else [
                        item
                        for item in entries.get("hostctl.shell_flavours", ())
                        if item.name == value
                    ]
                )
                if selected:
                    loaded = selected[0].load()
                    register_shell_flavour(loaded, name=value)
                    return _SHELLS[value]
            except Exception as plugin_error:
                raise ValueError(
                    f"invalid shell flavour plugin {value!r}: {plugin_error}"
                ) from plugin_error
        supported = ", ".join(sorted(_SHELLS))
        raise ValueError(
            f"unsupported shell flavour; choose one of: {supported}"
        ) from exc

Serial consoles

hostctl.SerialConsoleProtocol

Bases: Protocol

Runtime contract implemented by serial console profiles.

hostctl.RawConsoleProfile

Raw byte console; it supports interactive sessions but not run.

hostctl.PromptConsoleProfile(prompt, *, login=(), line_terminator=b'\r\n', error_patterns=(), status_marker=None, reliable_status=False, max_buffer=64 * 1024, read_size=4096, wakeup=b'\r', echo=True, paging_prompt=None, paging_continue=b' ', paging_disable=None, max_paging_pages=32, terminal_setup=None, status_parser=None)

Configurable prompt/login framing for terminal-like devices.

A profile only advertises run when reliable_status is explicitly enabled. Without a real status marker a prompt can delimit output, but it cannot truthfully represent a process exit status.

Source code in src/hostctl/serial/__init__.py

def __init__(
    self,
    prompt: bytes | str,
    *,
    login: typing.Iterable[LoginStep | tuple[bytes | str, bytes | str]] = (),
    line_terminator: bytes | str = b"\r\n",
    error_patterns: typing.Iterable[bytes | str] = (),
    status_marker: bytes | str | None = None,
    reliable_status: bool = False,
    max_buffer: int = 64 * 1024,
    read_size: int = 4096,
    wakeup: bytes = b"\r",
    echo: bool = True,
    paging_prompt: bytes | str | None = None,
    paging_continue: bytes | str = b" ",
    paging_disable: bytes | str | None = None,
    max_paging_pages: int = 32,
    terminal_setup: typing.Callable[[SerialProcess, int, int], None] | None = None,
    status_parser: typing.Callable[[re.Match[bytes]], int] | None = None,
) -> None:
    self._prompt = self._compile(prompt)
    self.login = tuple(
        step if isinstance(step, LoginStep) else LoginStep(step[0], step[1])
        for step in login
    )
    self._login_patterns = tuple(self._compile(step.expect) for step in self.login)
    self.line_terminator = (
        line_terminator.encode()
        if isinstance(line_terminator, str)
        else bytes(line_terminator)
    )
    if not self.line_terminator:
        raise ValueError("line_terminator must not be empty")
    self.error_patterns = tuple(self._compile(value) for value in error_patterns)
    self.status_marker = (
        self._compile(status_marker) if status_marker is not None else None
    )
    self.reliable_status = bool(reliable_status)
    self.can_run = self.reliable_status and self.status_marker is not None
    self.max_buffer = max(1024, int(max_buffer))
    self.read_size = max(1, int(read_size))
    self.wakeup = bytes(wakeup)
    self.echo = bool(echo)
    self._paging_prompt = self._compile(paging_prompt) if paging_prompt else None
    self.paging_continue = (
        paging_continue.encode()
        if isinstance(paging_continue, str)
        else bytes(paging_continue)
    )
    self.paging_disable = (
        paging_disable.encode()
        if isinstance(paging_disable, str)
        else (bytes(paging_disable) if paging_disable is not None else None)
    )
    if max_paging_pages < 0:
        raise ValueError("max_paging_pages must not be negative")
    self.max_paging_pages = int(max_paging_pages)
    self.terminal_setup = terminal_setup
    self.status_parser = status_parser
    self.encoding = "utf-8"

hostctl.LoginStep(expect, send, secret=False) dataclass

One bounded expect/send step used by :class:PromptConsoleProfile.

hostctl.ConsoleProtocolError

Bases: ConnectionError

The console did not complete the expected protocol exchange.

Provider composition

hostctl.ExecutorProvider(name, executor, *, capabilities=None, probe=None)

Named callable executor with a conservative probe hook.

Source code in src/hostctl/provider/_common.py

def __init__(
    self,
    name: str,
    executor: typing.Callable[..., object],
    *,
    capabilities=None,
    probe=None,
):
    if not name:
        raise ValueError("provider name must not be empty")
    self.name = name
    self.executor = executor
    raw = (
        getattr(executor, "executor_capabilities", ())
        if capabilities is None
        else capabilities
    )
    self.capabilities = frozenset(getattr(item, "value", item) for item in raw)
    self._probe = probe

hostctl.PathProvider(name, factory, *, capabilities=None, probe=None)

Named path factory. path must return a pathlib_next.Path.

Source code in src/hostctl/provider/_common.py

def __init__(
    self,
    name: str,
    factory: typing.Callable[..., Path],
    *,
    capabilities=None,
    probe=None,
):
    if not name:
        raise ValueError("provider name must not be empty")
    self.name = name
    self.factory = factory
    raw_capabilities = (
        self.DEFAULT_CAPABILITIES if capabilities is None else capabilities
    )
    self.capabilities = frozenset(
        getattr(item, "value", item) for item in raw_capabilities
    )
    self._probe = probe

hostctl.ProviderProbe(availability, reason='', capabilities=frozenset(), system_hint=None) dataclass

hostctl.ProviderSelection(provider, trace, generation=0, policy='ordered', pinned=False) dataclass

hostctl.ProviderSelector(providers=())

Ordered selector which never retries a dispatched operation.

Source code in src/hostctl/provider/_common.py

def __init__(self, providers=()):
    self.providers = tuple(providers)
    if any(not getattr(p, "name", "") for p in self.providers):
        raise ValueError("providers must have names")
    names = tuple(provider.name for provider in self.providers)
    if len(set(names)) != len(names):
        raise ValueError("provider names must be unique within a selector")
    self.last_selection: ProviderSelection | None = None
    self._probe_cache: dict[str, ProviderProbe] = {}
    self._declined: dict[str, str] = {}
    self._generation = 0
    # Trace entries accumulated across the failover attempts of one
    # caller-level operation, keyed by provider name so a later, more
    # informative record (a decline) supersedes the earlier optimistic one
    # (chosen).  See `select` for how an attempt sequence is delimited.
    self._attempt_trace: dict[str, dict[str, object]] = {}

declines property

Providers that refused before dispatch this generation, by name.

Maps provider name to the redacted refusal reason. A copy: mutating the result does not change selector state.

generation property

The current connection generation; probes are cached per value.

trace property

The accumulated trace for the operation currently being selected.

decline(name, reason='declined before dispatch')

Record that a provider refused before dispatch this generation.

A provider which raised OperationNotStarted proved that nothing started, so it is safe to skip it for the rest of this generation rather than re-attempting it on every later operation. The record is cleared by :meth:invalidate along with the probe cache.

Source code in src/hostctl/provider/_common.py

def decline(self, name: str, reason: str = "declined before dispatch") -> None:
    """Record that a provider refused *before* dispatch this generation.

    A provider which raised ``OperationNotStarted`` proved that nothing
    started, so it is safe to skip it for the rest of this generation
    rather than re-attempting it on every later operation.  The record is
    cleared by :meth:`invalidate` along with the probe cache.
    """
    self._declined[str(name)] = str(reason)
    log.debug(
        "provider %s declined before dispatch (generation %d): %s",
        self._safe_name(name),
        self._generation,
        self._safe_name(reason),
    )

invalidate()

Start a new generation, dropping cached probes and declines.

Source code in src/hostctl/provider/_common.py

def invalidate(self) -> None:
    """Start a new generation, dropping cached probes and declines."""
    self.last_selection = None
    self._probe_cache.clear()
    self._declined.clear()
    self._attempt_trace = {}
    self._generation += 1
    log.debug("selector invalidated; generation is now %d", self._generation)

probe(provider)

Return a provider probe cached for this selector generation.

A provider that declined before dispatch this generation reports as unavailable with the refusal reason, overriding the cached probe. The probe describes whether the transport could work; the decline is the newer and stronger evidence that right now it does not, so any caller reading availability (provider_details, capabilities) sees the refusal instead of a stale available.

Source code in src/hostctl/provider/_common.py

def probe(self, provider) -> ProviderProbe:
    """Return a provider probe cached for this selector generation.

    A provider that declined before dispatch this generation reports as
    ``unavailable`` with the refusal reason, overriding the cached probe.
    The probe describes whether the transport *could* work; the decline is
    the newer and stronger evidence that right now it does not, so any
    caller reading availability (``provider_details``, ``capabilities``)
    sees the refusal instead of a stale ``available``.
    """
    name = provider.name
    declined = self._declined.get(name)
    if declined is not None:
        return ProviderProbe("unavailable", self._safe_name(declined))
    if name in self._probe_cache:
        return self._probe_cache[name]
    try:
        result = provider.probe()
    except Exception as exc:
        result = ProviderProbe("unavailable", type(exc).__name__)
        log.debug(
            "provider %s probe raised %s; treating as unavailable",
            self._safe_name(name),
            type(exc).__name__,
        )
    self._probe_cache[name] = result
    return result

redact(value) classmethod

Return a diagnostic string with credential-like values removed.

Best effort, and deliberately so. This recognizes credentials that announce themselves -- a password=/token=/api_key= assignment, or the userinfo half of a scheme://user:secret@host URI. A bare positional secret (mysql -p hunter2, an argv element that is only a token) carries no marker and cannot be detected by inspection. Treat debug output as sensitive.

Source code in src/hostctl/provider/_common.py

@classmethod
def redact(cls, value: object) -> str:
    """Return a diagnostic string with credential-like values removed.

    Best effort, and deliberately so.  This recognizes credentials that
    *announce themselves* -- a ``password=``/``token=``/``api_key=``
    assignment, or the userinfo half of a ``scheme://user:secret@host``
    URI.  A bare positional secret (``mysql -p hunter2``, an argv element
    that is only a token) carries no marker and cannot be detected by
    inspection.  Treat debug output as sensitive.
    """
    return cls._safe_name(value)

hostctl.OperationNotStarted(reason='operation was not started', *, cause=None)

Bases: RuntimeError

A provider declined before a remote operation could start.

Source code in src/hostctl/provider/_common.py

def __init__(
    self,
    reason: str = "operation was not started",
    *,
    cause: Exception | None = None,
):
    super().__init__(reason)
    self.reason = reason
    self.cause = cause

hostctl.SessionInitializer(initialize, timeout=None) dataclass

Optional post-connect bootstrap hook receiving the connected system host.

hostctl.register_system_provider(name, resolver)

Register a provider descriptor resolver for :class:SystemConfig.

Source code in src/hostctl/host/system.py

def register_system_provider(name: str, resolver) -> None:
    """Register a provider descriptor resolver for :class:`SystemConfig`."""
    key = str(name).casefold().strip()
    if not key or not callable(resolver):
        raise ValueError("provider name and callable resolver are required")
    _SYSTEM_PROVIDER_RESOLVERS[key] = resolver

hostctl.CompositePosixPath(*segments, **kwargs)

Bases: _CompositePathMixin, PosixPathname, Path

A provider-selecting path with POSIX syntax on every client OS.

Source code in src/hostctl/host/composite_path.py

def __init__(self, *segments, **kwargs):
    if not hasattr(self, "_raw_paths") and not hasattr(self, "_parts"):
        pathlib.PurePath.__init__(self, *segments)

hostctl.CompositeWindowsPath(*segments, **kwargs)

Bases: _CompositePathMixin, WindowsPathname, Path

A provider-selecting path with Windows syntax on every client OS.

Source code in src/hostctl/host/composite_path.py

def __init__(self, *segments, **kwargs):
    if not hasattr(self, "_raw_paths") and not hasattr(self, "_parts"):
        pathlib.PurePath.__init__(self, *segments)

Connection strings

hostctl.ConnectionString(value, *, scheme=None, host=None, port=None, username=None, password=None, extras=None, path=None, query=None, fragment=None, defaults=None) dataclass

A connection target parsed from whatever a user typed.

ConnectionString("nas"), ConnectionString("nas:8443"), and ConnectionString("wss://root:pw@nas:8443/api") all parse. A bare host is not an invalid URI -- it is a URI with the scheme left off, which is what people type on a command line.

Every field can also be supplied directly, and three layers decide each one: an explicit argument wins, then whatever the string carried, then defaults.

So scheme=, port=, and the rest are overrides, not defaults -- ConnectionString("wss://nas", scheme="ssh") is ssh, the same way port= beats a port written in the string. Supplying a value only when the string omits one is exactly what defaults is for:

ConnectionString("nas", scheme="wss")               # wss://nas
ConnectionString("wss://nas", scheme="ssh")         # ssh://nas
ConnectionString("wss://nas", defaults={"scheme": "ssh"})   # wss://nas
ConnectionString("nas:9", scheme="wss", port=1)     # port=1 beats :9

defaults accepts a mapping or another ConnectionString, so a partially filled one expresses "these settings unless the string says otherwise":

profile = ConnectionString("wss://root@nas", port=443)
ConnectionString("other", defaults=profile)      # wss://root@other:443

Only fields a ConnectionString actually carries count as defaults, so an empty path/query/fragment never overrides.

A port carries its own resolution strategy, wherever one is accepted -- as port=, in defaults, or written in the string. It may be:

• an int, the port itself;
• a callable scheme -> port | None, asked once the scheme is settled;
• anything indexable by scheme (a plain mapping included), so a caller passes its whole table without pre-selecting an entry;
• None, meaning nothing was supplied here, so the next layer decides;
• False, meaning no port and no lookup -- stop here.

When no layer supplies one, netimps.get_default_port resolves the scheme. It knows schemes the system services database does not (ws/wss, socks5), and an application can register more with netimps.register_port.

ConnectionString("nas", scheme="wss")                       # :443
ConnectionString("nas", scheme="wss", port=8443)            # :8443
ConnectionString("nas", scheme="wss", port={"wss": 9})       # :9
ConnectionString("nas", scheme="wss", port=lambda s: 9)      # :9
ConnectionString("nas", scheme="wss", port=False)            # no port

A scheme a table or callable does not know is not an error -- it means "no conventional port for this one", and the netimps fallback then applies. A resolver that answers False declines that fallback, the same as passing False directly.

The host keeps the spelling it was given. urlsplit().hostname is case-folded, which is right for resolution and wrong for text rendered back to a caller, so nasA stays nasA in host and in str(). DNS treats the two as one name, so nothing about reaching the target changes.

Credentials never render. str() and repr() both emit the redacted form -- the password is removed rather than masked, so the output stays a valid, reusable connection string that cannot round-trip a wrong credential. password is excluded from the generated repr, so a value reaching a log line or a traceback frame does not leak it.

A password may carry credential extras after a newline, exactly as :func:parse_credentials describes; extras holds them and password is just the password. The separator may be written raw, since characters urlsplit would silently delete are percent-encoded before parsing.

Source code in src/hostctl/host/_connection.py

def __init__(
    self,
    value: _ty.Union[str, "ConnectionString"],
    *,
    scheme: _ty.Optional[str] = None,
    host: _ty.Optional[str] = None,
    port: "Port" = None,
    username: _ty.Optional[str] = None,
    password: _ty.Optional[str] = None,
    extras: _ty.Optional[_ty.Mapping[str, str]] = None,
    path: _ty.Optional[str] = None,
    query: _ty.Optional[str] = None,
    fragment: _ty.Optional[str] = None,
    defaults: _ty.Union["ConnectionString", _ty.Mapping[str, object], None] = None,
) -> None:
    overrides = {
        "scheme": scheme,
        "host": host,
        "port": port,
        "username": username,
        "password": password,
        "extras": extras,
        "path": path,
        "query": query,
        "fragment": fragment,
    }
    fallbacks = _as_defaults(defaults)
    unknown = set(fallbacks) - {field.name for field in _dc.fields(self)}
    if unknown:
        raise TypeError(f"unknown default field: {sorted(unknown)[0]}")

    if isinstance(value, ConnectionString):
        parsed_values: _ty.Dict[str, object] = {
            field.name: getattr(value, field.name) for field in _dc.fields(self)
        }
    else:
        # A scheme may come from the explicit argument or the defaults;
        # either lets a bare host parse, so resolve it before splitting.
        assumed = scheme or fallbacks.get("scheme")
        parsed = _split(value, _ty.cast(_ty.Optional[str], assumed))
        _reject_authority_control_characters(parsed)
        parsed_password, parsed_extras = None, {}
        if parsed.password is not None:
            parsed_password, parsed_extras = parse_credentials(
                _unquote(parsed.password)
            )
        parsed_values = {
            "scheme": parsed.scheme.casefold() or None,
            "host": uri_hostname(parsed) or None,
            "port": parsed.port,
            "username": _unquote(parsed.username) if parsed.username else None,
            "password": parsed_password,
            "extras": dict(parsed_extras) or None,
            "path": parsed.path or None,
            "query": parsed.query or None,
            "fragment": parsed.fragment or None,
        }

    # Precedence: an explicit argument wins, then what the string carried,
    # then `defaults`. A default therefore fills a gap rather than
    # overriding something the caller actually wrote.
    resolved: _ty.Dict[str, object] = {}
    for name, override in overrides.items():
        for candidate in (override, parsed_values.get(name), fallbacks.get(name)):
            if candidate is not None:
                resolved[name] = candidate
                break
        else:
            resolved[name] = None

    if resolved["scheme"] is None:
        raise ValueError(
            f"connection string has no scheme: {value!r}; pass scheme= to set "
            "one, or defaults= to supply it only when the string omits it"
        )
    scheme_text = _ty.cast(str, resolved["scheme"]).casefold()
    # Only now is the scheme settled, so a callable or table given for the
    # port can be asked. When no layer supplied anything, netimps resolves
    # the scheme -- it knows ones the system services database does not
    # (`ws`/`wss`, `socks5`), and an application can add more with
    # `netimps.register_port`.
    port_value = _resolve_port(scheme_text, _ty.cast("Port", resolved["port"]))
    if port_value is _NO_PORT:
        port_value = None
    elif port_value is None:
        port_value = _get_default_port(scheme_text)
    resolved["port"] = port_value

    object.__setattr__(self, "scheme", scheme_text)
    object.__setattr__(self, "host", resolved["host"] or "")
    object.__setattr__(self, "port", resolved["port"])
    object.__setattr__(self, "username", resolved["username"])
    object.__setattr__(self, "password", resolved["password"])
    object.__setattr__(self, "extras", dict(resolved["extras"] or {}))
    object.__setattr__(self, "path", resolved["path"] or "")
    object.__setattr__(self, "query", resolved["query"] or "")
    object.__setattr__(self, "fragment", resolved["fragment"] or "")

authority property

The authority as it renders, without any password.

is_local property

Whether this names the local machine, without resolving anything.

Deliberately does no I/O: this is called while a configuration is built, and that is network-free. Resolving would block, and would raise on a name that does not resolve -- so it could not even be used defensively.

An address literal is answered by netimps.is_local_address, which covers loopback and addresses actually assigned to this machine -- 10.0.0.5 is local when it is one of this host's own. A name is only compared against the loopback spellings, since deciding anything more about a name requires a lookup.

qsl property

Query pairs in order, keeping repeats.

geturl()

The credential-free connection string, valid and reusable.

Source code in src/hostctl/host/_connection.py

def geturl(self) -> str:
    """The credential-free connection string, valid and reusable."""
    return _SplitResult(
        self.scheme, self.authority, self.path, self.query, self.fragment
    ).geturl()

query_val(name, default=None)

The last value for name, or default. Last wins, as in a URI.

Source code in src/hostctl/host/_connection.py

def query_val(self, name: str, default: _ty.Optional[str] = None):
    """The last value for `name`, or `default`. Last wins, as in a URI."""
    found = default
    for key, value in self.qsl:
        if key == name:
            found = value
    return found

replace(**changes)

A copy with changes applied.

Built field by field rather than through dataclasses.replace, which would re-invoke __init__ -- and __init__ parses a string, so it does not accept the field names.

Source code in src/hostctl/host/_connection.py

def replace(self, **changes: object) -> "ConnectionString":
    """A copy with `changes` applied.

    Built field by field rather than through `dataclasses.replace`, which
    would re-invoke `__init__` -- and `__init__` parses a string, so it
    does not accept the field names.
    """
    names = {field.name for field in _dc.fields(self)}
    unknown = set(changes) - names
    if unknown:
        raise TypeError(f"unknown field: {sorted(unknown)[0]}")
    copy = object.__new__(type(self))
    for name in names:
        object.__setattr__(copy, name, changes.get(name, getattr(self, name)))
    return copy

hostctl.redact_uri(uri)

Strip any password from uri, leaving a valid, reusable URI.

scheme://user:secret@host becomes scheme://user@host. The password is removed rather than masked: a placeholder would make the URI round-trip into a wrong credential if anything fed the rendered form back in, and would not be a real password anyway. What comes out is the same canonical, credential-free string a config renders, so it is safe to log, repr, or hand back to HostConfig.

A URI with no password is returned unchanged.

This never raises. It is meant for error messages and log records, where failing to render a diagnostic would be worse than rendering an odd one. Characters urlsplit would delete (tab, CR, LF) are percent-encoded first, the same as during dispatch, so a password written with a raw newline is still recognized and removed rather than partly surviving into the output.

Source code in src/hostctl/host/_common.py

def redact_uri(uri: str) -> str:
    """Strip any password from `uri`, leaving a valid, reusable URI.

    `scheme://user:secret@host` becomes `scheme://user@host`. The password is
    removed rather than masked: a placeholder would make the URI round-trip
    into a *wrong* credential if anything fed the rendered form back in, and
    would not be a real password anyway. What comes out is the same canonical,
    credential-free string a config renders, so it is safe to log, repr, or
    hand back to `HostConfig`.

    A URI with no password is returned unchanged.

    This never raises. It is meant for error messages and log records, where
    failing to render a diagnostic would be worse than rendering an odd one.
    Characters `urlsplit` would delete (tab, CR, LF) are percent-encoded first,
    the same as during dispatch, so a password written with a raw newline is
    still recognized and removed rather than partly surviving into the output.
    """
    parsed = _urlsplit(_encode_stripped_characters(uri))
    if parsed.password is None:
        return uri
    return _without_password(parsed).geturl()

hostctl.parse_credentials(password)

Split a password field into the password and any trailing extras.

A newline separates the password from additional credential values, one per line, each key:value:

"hunter2"                       -> ("hunter2", {})
"hunter2\notp:123456"           -> ("hunter2", {"otp": "123456"})
"hunter2\notp:123456\nrealm:CORP" -> ("hunter2", {"otp": ..., "realm": ...})

A bare name with no : is a flag -- it maps to the empty string, exactly as name: does:

"hunter2\ninteractive"           -> ("hunter2", {"interactive": ""})

This is how a second factor reaches a transport through a single password field -- a URI's userinfo, an environment variable, a prompt -- without every caller inventing its own encoding. A newline is assumed not to be a valid password character, which is the same assumption pytruenas makes.

Names are casefolded and stripped of surrounding whitespace. A value is everything after the first :, taken verbatim -- it may contain colons, and its leading and trailing whitespace is preserved, because a secret may legitimately begin or end with a space and silently trimming one would fail authentication with no visible cause. Blank lines are ignored.

Source code in src/hostctl/host/_common.py

def parse_credentials(password: str) -> _ty.Tuple[str, _ty.Dict[str, str]]:
    """Split a password field into the password and any trailing extras.

    A newline separates the password from additional credential values, one
    per line, each `key:value`:

        "hunter2"                       -> ("hunter2", {})
        "hunter2\\notp:123456"           -> ("hunter2", {"otp": "123456"})
        "hunter2\\notp:123456\\nrealm:CORP" -> ("hunter2", {"otp": ..., "realm": ...})

    A bare name with no `:` is a flag -- it maps to the empty string, exactly
    as `name:` does:

        "hunter2\\ninteractive"           -> ("hunter2", {"interactive": ""})

    This is how a second factor reaches a transport through a single password
    field -- a URI's userinfo, an environment variable, a prompt -- without
    every caller inventing its own encoding. A newline is assumed not to be a
    valid password character, which is the same assumption pytruenas makes.

    Names are casefolded and stripped of surrounding whitespace. A value is
    everything after the first `:`, taken **verbatim** -- it may contain
    colons, and its leading and trailing whitespace is preserved, because a
    secret may legitimately begin or end with a space and silently trimming
    one would fail authentication with no visible cause. Blank lines are
    ignored.
    """
    # Split on any line ending: a value pasted from a file or a Windows prompt
    # arrives CRLF-terminated, and a trailing "\r" left on the password would
    # fail authentication with no visible cause.
    lines = password.splitlines()
    if len(lines) <= 1:
        return password, {}
    password, remainder = lines[0], lines[1:]
    extras: _ty.Dict[str, str] = {}
    for line in remainder:
        if not line.strip():
            continue
        # A line with no ":" is a bare flag; `partition` already yields "" for
        # its value, so flags and `name:` need no separate branch.
        key, _, value = line.partition(":")
        key = key.strip().casefold()
        if not key:
            raise ValueError("credential extra names must not be empty")
        extras[key] = value
    return password, extras

hostctl.strict_uri_credentials(credentials, allowed)

Reject credentials which the selected implementation does not accept.

Source code in src/hostctl/host/_common.py

def strict_uri_credentials(
    credentials: _ty.Mapping[str, object], allowed: _ty.Iterable[str]
) -> None:
    """Reject credentials which the selected implementation does not accept."""
    unknown = set(credentials) - set(allowed)
    if unknown:
        raise ValueError(f"unknown credential argument: {sorted(unknown)[0]}")

hostctl.strict_uri_query(parsed, allowed)

Parse one selected implementation's query without ambiguity.

Source code in src/hostctl/host/_common.py

def strict_uri_query(
    parsed: _SplitResult, allowed: _ty.Iterable[str]
) -> _ty.Dict[str, str]:
    """Parse one selected implementation's query without ambiguity."""
    query = {}
    for key, value in _parse_qsl(parsed.query, keep_blank_values=True):
        if key in query:
            raise ValueError(f"duplicate connection parameter: {key}")
        query[key] = value
    unknown = set(query) - set(allowed)
    if unknown:
        raise ValueError(f"unknown connection parameter: {sorted(unknown)[0]}")
    return query

hostctl.uri_host(host)

Bracket an IPv6 literal for use in a URI authority.

Source code in src/hostctl/host/_common.py

def uri_host(host: str) -> str:
    """Bracket an IPv6 literal for use in a URI authority."""
    return f"[{host}]" if ":" in host and not host.startswith("[") else host

hostctl.uri_hostname(parsed)

The host as it was written, not as urlsplit case-folded it.

urlsplit().hostname is lowercased by design -- DNS is case-insensitive, so folding is right for resolution. It is wrong for text a caller will see again. A config built from a URI stores this value and renders it back through connection_uri, so reading .hostname there would make the library echo a spelling the operator never typed, and nasA would reach logs and HostInfo as nasa.

Use this in _from_parsed_uri wherever a host is stored. Presence checks (if not parsed.hostname) can keep using .hostname -- emptiness does not depend on case. Resolution is unaffected either way: DNS, SSH, and WinRM all treat the two spellings as one name.

Returns "" for a URI with no host, matching hostname or "".

hostname is a case-folded substring of the authority (after any userinfo), so locating it recovers the original text.

Source code in src/hostctl/host/_common.py

def uri_hostname(parsed: _SplitResult) -> str:
    """The host as it was written, not as `urlsplit` case-folded it.

    `urlsplit().hostname` is lowercased by design -- DNS is case-insensitive,
    so folding is right for *resolution*. It is wrong for text a caller will
    see again. A config built from a URI stores this value and renders it back
    through `connection_uri`, so reading `.hostname` there would make the
    library echo a spelling the operator never typed, and `nasA` would reach
    logs and `HostInfo` as `nasa`.

    Use this in `_from_parsed_uri` wherever a host is stored. Presence checks
    (`if not parsed.hostname`) can keep using `.hostname` -- emptiness does not
    depend on case. Resolution is unaffected either way: DNS, SSH, and WinRM
    all treat the two spellings as one name.

    Returns `""` for a URI with no host, matching `hostname or ""`.

    `hostname` is a case-folded substring of the authority (after any
    userinfo), so locating it recovers the original text.
    """
    hostname = parsed.hostname or ""
    authority = parsed.netloc.rpartition("@")[2]
    index = authority.casefold().find(hostname)
    if index < 0:
        return hostname
    return authority[index : index + len(hostname)]

Composing transports

hostctl.ssh_providers(config)

Build an executor and path provider sharing one SSH transport.

This is the supported way to compose SSH into a host you assemble yourself, rather than taking the finished PosixHost that SshConfig._create_host() returns:

executors, paths = [], []
run_provider, path_provider = ssh_providers(config.ssh)
executors.append(run_provider)
paths.append(path_provider)

Both providers must share one transport: that is what makes them share a connection and a lifecycle, and SshExecutorProvider is the one that owns close. Assembling them by hand from two transports type-checks and silently opens two connections, only one of which is ever closed -- which is precisely why this returns a pair instead of exposing the transport.

Source code in src/hostctl/host/_ssh.py

def ssh_providers(
    config: SshConfig,
) -> typing.Tuple[SshExecutorProvider, SftpPathProvider]:
    """Build an executor and path provider sharing one SSH transport.

    This is the supported way to compose SSH into a host you assemble
    yourself, rather than taking the finished `PosixHost` that
    `SshConfig._create_host()` returns:

        executors, paths = [], []
        run_provider, path_provider = ssh_providers(config.ssh)
        executors.append(run_provider)
        paths.append(path_provider)

    Both providers **must** share one transport: that is what makes them share
    a connection and a lifecycle, and `SshExecutorProvider` is the one that
    owns close. Assembling them by hand from two transports type-checks and
    silently opens two connections, only one of which is ever closed -- which
    is precisely why this returns a pair instead of exposing the transport.
    """
    transport = _SshTransport(config)
    return SshExecutorProvider(transport), SftpPathProvider(transport)

hostctl.winrm_providers(config)

Build an executor and path provider sharing one WinRM transport.

The SSH counterpart, hostctl.host.ssh_providers, documents why this returns a pair rather than exposing the transport: both providers must share one, and building them separately silently opens two.

Source code in src/hostctl/host/_winrm.py

def winrm_providers(
    config: WinRMConfig,
) -> typing.Tuple[WinRMExecutorProvider, WinRMPathProvider]:
    """Build an executor and path provider sharing one WinRM transport.

    The SSH counterpart, `hostctl.host.ssh_providers`, documents why this
    returns a pair rather than exposing the transport: both providers must
    share one, and building them separately silently opens two.
    """
    transport = _WinRMTransport(config)
    return WinRMExecutorProvider(transport), WinRMPathProvider(transport)

Transfers and checksums

hostctl.ProgressReader(reader, callback, *, total=None)

Wrap a binary reader and report (bytes_read, total_bytes).

Source code in src/hostctl/sync.py

def __init__(
    self,
    reader: typing.BinaryIO,
    callback: typing.Callable[[int, typing.Optional[int]], None],
    *,
    total: typing.Optional[int] = None,
) -> None:
    if not callable(callback):
        raise TypeError("callback must be callable")
    self.reader = reader
    self.callback = callback
    self.total = total
    self.bytes_read = 0

hostctl.host_checksum(*hosts, algorithm='md5', chunk_size=1024 * 1024)

Build a PathSyncer checksum using execution beside owned paths.

More than one host may be supplied for a cross-host sync. Paths which do not belong to any supplied host are hashed through their binary open() contract, preserving interoperability with arbitrary pathlib_next implementations.

Source code in src/hostctl/sync.py

def host_checksum(
    *hosts: Host,
    algorithm: str = "md5",
    chunk_size: int = 1024 * 1024,
) -> typing.Callable[[PathAndStat], str]:
    """Build a ``PathSyncer`` checksum using execution beside owned paths.

    More than one host may be supplied for a cross-host sync. Paths which do
    not belong to any supplied host are hashed through their binary ``open()``
    contract, preserving interoperability with arbitrary ``pathlib_next``
    implementations.
    """
    normalized = algorithm.casefold().replace("-", "")
    if normalized not in _REMOTE_ALGORITHMS:
        raise ValueError(
            f"unsupported remote checksum algorithm: {algorithm!r}; "
            f"choose one of {', '.join(sorted(_REMOTE_ALGORITHMS))}"
        )
    if not hosts:
        raise ValueError("host_checksum requires at least one host")
    if chunk_size <= 0:
        raise ValueError("chunk_size must be positive")

    owners = tuple((host, _host_path_token(host)) for host in hosts)

    def checksum(entry: PathAndStat) -> str:
        for host, owner_token in owners:
            if _host_owns_path(host, entry.path, owner_token):
                try:
                    return _remote_checksum(host, entry.path, normalized)
                except _UNAVAILABLE_REMOTE_TOOL:
                    # The host owns the path but cannot hash it in place: the
                    # tool is missing, refused, or answered unintelligibly.
                    # Reading the content is slower but still correct, so a
                    # sync degrades rather than aborting.
                    break
        return _stream_checksum(entry.path, normalized, chunk_size)

    return checksum

hostctl.stat_checksum(entry)

Return the cached size and modification time without reading content.

This is rsync's quick check: no content is read and no command is run. Two caveats decide whether it suits a given sync.

It can miss a change which preserves both size and modification time.

It can also report a change which is not one, on interpreters where pathlib_next.Path.copy() preserves st_mode but not timestamps: a file this helper copies lands with a fresh modification time and compares unequal on the next run, so the sync never settles. Python 3.14 added a stdlib Path.copy() which does preserve timestamps, and a local path resolves to it there, so the same sync converges. Because that difference is version- and backend-dependent, use this helper where source modification times are meaningful on both sides -- a tree replicated by something which preserves them -- and prefer :func:host_checksum when hostctl's own copies must converge to a no-op on every supported interpreter.

Source code in src/hostctl/sync.py

def stat_checksum(entry: PathAndStat) -> tuple[int, float]:
    """Return the cached size and modification time without reading content.

    This is rsync's quick check: no content is read and no command is run.
    Two caveats decide whether it suits a given sync.

    It can *miss* a change which preserves both size and modification time.

    It can also *report* a change which is not one, on interpreters where
    ``pathlib_next.Path.copy()`` preserves ``st_mode`` but not timestamps: a
    file this helper copies lands with a fresh modification time and compares
    unequal on the next run, so the sync never settles.  Python 3.14 added a
    stdlib ``Path.copy()`` which does preserve timestamps, and a local path
    resolves to it there, so the same sync converges.  Because that difference
    is version- and backend-dependent, use this helper where source
    modification times are meaningful on both sides -- a tree replicated by
    something which preserves them -- and prefer :func:`host_checksum` when
    hostctl's own copies must converge to a no-op on every supported
    interpreter.
    """
    if entry.stat is None:
        raise FileNotFoundError(entry.path)
    return entry.stat.st_size, entry.stat.st_mtime

Aliases and constants

HostPath is pathlib_next.Path — the type every usable host.path() returns, re-exported so callers can annotate against it without depending on pathlib_next by name.

ExecutorCommand (str | pathlib.PurePath | pathlib_next.Pathname) is the command type accepted by Executor and Shell.execute(). ExecutorCapability is the enum whose .ARGS, .CWD, and .ENV members declare native executor support. Its members subclass str, so ExecutorCapability.CWD == "cwd": capability sets are compared as strings throughout, which lets providers advertise transport-specific tokens ("runspace") alongside the named members in one vocabulary.

POSIX_SHELL and POWERSHELL are the built-in shell-flavour instances; BASH, ZSH, FISH, CMD, and PWSH are the other ready-to-use flavours. Any of them can be passed wherever a ShellFlavour is expected.

pypsrp_available() reports whether the optional PSRP extra is importable and require_pypsrp() raises an actionable ImportError naming the extra when it is not. __version__ contains the installed package version.