plesty.server.services

Services: the model’s side effects on the host.

Everything that touches the operating system — uv and git for installing, subprocesses for running servers and jobs — lives here. Services take model objects in and hand model objects back; presenters drive them and never call the tools directly, so a service is the one place a tool’s quirks are handled and the one seam tests replace.

Submodules

Attributes

CHECKOUT_VENV

PYTHON_SPEC

Echo

Runner

PROBE_HOST

Exceptions

GitUnavailable

No git on the bench, and none could be installed.

InstallError

uv/git failed, or the environment is not in the shape expected.

CatalogueError

The catalogue could not be fetched or read.

AlreadyRunning

start was asked for a device whose server is already up.

NotRunning

stop was asked for a device that has no live server.

PortBusy

start found the device's port already answering — another server holds it.

SupervisorError

A start or stop request could not be carried out.

Classes

Installer

Install, update and remove device environments; records them in the manager.

Agent

Serve plesty.bench clients for one bench.

Catalogue

The hub catalogue at one index URL.

JobRunner

Start, observe and cancel jobs under a Home.

Supervisor

Launch and control device servers under a Home.

Functions

bench_git(→ pathlib.Path)

Where the bench keeps its own git (whether or not one is there yet).

find_git(→ str | None)

A usable git: PATH first, then the usual installs, then the bench's own.

install_git(→ str)

Unpack a portable git into <home>/tools/git; returns the executable.

default_runner(→ subprocess.CompletedProcess[str])

Run cmd, streaming its merged output to echo as it arrives.

serve_in_thread(→ tuple[threading.Thread, threading.Event])

Run agent on a daemon thread; returns the thread and its stop event.

http_fetch(→ bytes)

GET url with the standard library (the bench may have no requests).

pypi_release(→ str | None)

The newest release of package on PyPI, or None when it has none there.

probe(→ bool)

Ask the server on port to describe itself.

request(→ dict[str, Any])

Send one request to the server on port and return its reply.

Package Contents

exception plesty.server.services.GitUnavailable

Bases: RuntimeError

No git on the bench, and none could be installed.

Initialize self. See help(type(self)) for accurate signature.

plesty.server.services.bench_git(home: plesty.server.model.home.Home) → pathlib.Path

Where the bench keeps its own git (whether or not one is there yet).

Parameters:

home (plesty.server.model.home.Home)

Return type:

pathlib.Path

plesty.server.services.find_git(home: plesty.server.model.home.Home) → str | None

A usable git: PATH first, then the usual installs, then the bench’s own.

Parameters:

home (plesty.server.model.home.Home) – The bench home, which may hold a git of its own.

Returns:

The executable, or None when the bench has none.

Return type:

str | None

plesty.server.services.install_git(home: plesty.server.model.home.Home, echo: Echo | None = None, fetch: Fetcher | None = None) → str

Unpack a portable git into <home>/tools/git; returns the executable.

Windows only: it is the platform where a bench routinely has no git and no way to install one. The archive is verified against the pinned digest before anything is unpacked — a mismatch is an error, not a warning.

Parameters:
Raises:

GitUnavailable – Not Windows, unknown architecture, download failed, digest mismatch, or the archive held no git.

Return type:

str

plesty.server.services.CHECKOUT_VENV = '.venv'
plesty.server.services.PYTHON_SPEC = '>=3.12'
plesty.server.services.Echo
class plesty.server.services.Installer(home: plesty.server.model.home.Home, manager: plesty.server.model.device_manager.DeviceManager | None = None, uv: str | None = None, git: str | None = None, runner: Runner | None = None, echo: Echo | None = None)

Install, update and remove device environments; records them in the manager.

Bind the installer to a home.

Parameters:

  • home (plesty.server.model.home.Home) – The bench home.

  • manager (plesty.server.model.device_manager.DeviceManager | None) – The inventory to record into; a fresh one on home by default.

  • uv (str | None) – Path of the uv executable; looked up on PATH by default.

  • git (str | None) – Path of the git executable; found (or installed) by default.

  • runner (Runner | None) – Executes a command line — injected by tests.

  • echo (Echo | None) – Where each line uv and git write goes while they run. Defaults to stdout, which is the job log when run as a job.

home
manager
_uv = None
_git = None
_run
_echo: Echo
property uv: str

The uv executable; raises InstallError when the bench has none.

Return type:

str

_find_uv() → str

PATH first, then where uv installs itself.

Raises:

InstallError – When there is no uv to be found.

Return type:

str

property git: str

The git executable, installing one for the bench when it has none.

Raises:

InstallError – When the bench has no git and cannot be given one.

Return type:

str

_on_path(git: str) → str

Put git’s directory on PATH for this process and its children.

Calling git by absolute path is enough for the installer’s own clone and checkout — and not enough for the install. A module’s dependencies may be git+https URLs, and the uv this spawns to resolve them looks for git on PATH and nowhere else: on the very bench this feature exists for, one with no git at all, the install would fetch a portable git, clone with it, and then fail in uv sync with “Git executable not found”. The same holds for a git found under %ProgramFiles% but absent from PATH.

Parameters:

git (str)

Return type:

str

_find_or_install_git() → str

Locate git, or fetch a portable one (Windows); explains itself either way.

Return type:

str

python(installed: plesty.server.model.device_manager.InstalledVersion) → pathlib.Path

The interpreter that runs installed.

Parameters:

installed (plesty.server.model.device_manager.InstalledVersion)

Return type:

pathlib.Path

install(spec: plesty.server.model.device.DeviceSpec, force: bool = False, job_id: str | None = None) → plesty.server.model.device_manager.InstalledVersion

Make sure an environment serving spec exists; returns its record.

Parameters:

  • spec (plesty.server.model.device.DeviceSpec) – What to install: package plus a pinned version (PyPI) or a repository and ref (git checkout).

  • force (bool) – Reinstall even when a satisfying environment is on record.

  • job_id (str | None) – The job this runs as, stored on the record.

Raises:

InstallError – When uv/git fail or the package did not end up installed.

Return type:

plesty.server.model.device_manager.InstalledVersion

_install_the_tag_instead(spec: plesty.server.model.device.DeviceSpec, job_id: str | None, failure: InstallError) → plesty.server.model.device_manager.InstalledVersion

After a failed release install, install the same version from the repository.

A module is on the hub before it is on PyPI: the registry knows the release from its tag, and the package reaches the index only once the release job has run — so “that release is not on the index” is a routine state of a young module, not a broken bench. The repository holds the same version under v<version>, which is the ref a spec with a version already resolves to.

Raises:

InstallError – When no repository is known for the package, or the repository has no such tag either — carrying both reasons.

Parameters:
Return type:

plesty.server.model.device_manager.InstalledVersion

update(installed: plesty.server.model.device_manager.InstalledVersion, ref: str | None = None, job_id: str | None = None) → plesty.server.model.device_manager.InstalledVersion

Bring installed up to date; returns the (possibly new) record.

Git checkout: fetch, check out ref (default: the recorded ref), fast-forward, uv sync. PyPI: reinstall the same version.

Parameters:
Return type:

plesty.server.model.device_manager.InstalledVersion

uninstall(installed: plesty.server.model.device_manager.InstalledVersion) → bool

Delete the environment of installed and forget it.

Returns:

True when a directory was removed.

Parameters:

installed (plesty.server.model.device_manager.InstalledVersion)

Return type:

bool

_install_release(spec: plesty.server.model.device.DeviceSpec, job_id: str | None) → plesty.server.model.device_manager.InstalledVersion
Parameters:
Return type:

plesty.server.model.device_manager.InstalledVersion

_install_checkout(spec: plesty.server.model.device.DeviceSpec, job_id: str | None) → plesty.server.model.device_manager.InstalledVersion
Parameters:
Return type:

plesty.server.model.device_manager.InstalledVersion

_uv_call(args: list[str], what: str, quiet: bool = False) → subprocess.CompletedProcess[str]
Parameters:

  • args (list[str])

  • what (str)

  • quiet (bool)

Return type:

subprocess.CompletedProcess[str]

_uv_resolve(args: list[str], what: str) → subprocess.CompletedProcess[str]

Run a resolving uv command, retrying once with pre-releases allowed.

A hub module under development depends on a .dev release of plesty-lib, and uv considers only stable versions unless it is told otherwise — it fails with a hint naming the pre-release. Allowing them from the start would pull a pre-release into every install that has a stable answer, so it is the retry, not the rule.

Parameters:

  • args (list[str])

  • what (str)

Return type:

subprocess.CompletedProcess[str]

_git_call(args: list[str], what: str, quiet: bool = False) → subprocess.CompletedProcess[str]
Parameters:

  • args (list[str])

  • what (str)

  • quiet (bool)

Return type:

subprocess.CompletedProcess[str]

_call(cmd: list[str], what: str, quiet: bool = False) → subprocess.CompletedProcess[str]

Run cmd; quiet keeps a query’s output out of the log (it is read, not watched).

Parameters:

  • cmd (list[str])

  • what (str)

  • quiet (bool)

Return type:

subprocess.CompletedProcess[str]

_installed_version(python: pathlib.Path, package: str) → str | None
Parameters:

  • python (pathlib.Path)

  • package (str)

Return type:

str | None

_python_version(python: pathlib.Path) → str
Parameters:

python (pathlib.Path)

Return type:

str

exception plesty.server.services.InstallError

Bases: RuntimeError

uv/git failed, or the environment is not in the shape expected.

Initialize self. See help(type(self)) for accurate signature.

plesty.server.services.Runner
plesty.server.services.default_runner(cmd: list[str], echo: Echo | None = None) → subprocess.CompletedProcess[str]

Run cmd, streaming its merged output to echo as it arrives.

stderr is folded into stdout so the order the tool wrote them in survives; the text is returned as well, so a failure still carries its own detail.

Parameters:

  • cmd (list[str])

  • echo (Echo | None)

Return type:

subprocess.CompletedProcess[str]

class plesty.server.services.Agent(home: plesty.server.model.home.Home, fleet_path: str | os.PathLike[str] | None = None, bind: str = DEFAULT_BIND, token: str | None = None, workers: int = 4)

Serve plesty.bench clients for one bench.

Configure the agent.

Parameters:

  • home (plesty.server.model.home.Home) – The bench home every command runs against.

  • fleet_path (str | os.PathLike[str] | None) – Fleet file passed to every command (-f), or None.

  • bind (str) – ZMQ endpoint to bind.

  • token (str | None) – Shared secret; None reads PLESTY_BENCH_TOKEN; empty means no check (a bench on a trusted network).

  • workers (int) – Concurrent requests.

home
fleet_path
bind = 'tcp://*:5550'
token
workers = 4
bound: str | None = None
served = 0
handle(message: Any) → dict[str, Any]

Answer one decoded request (pure; no socket involved).

Parameters:

message (Any)

Return type:

dict[str, Any]

description() → dict[str, Any]

What describe returns.

Return type:

dict[str, Any]

exec(argv: collections.abc.Sequence[Any]) → dict[str, Any]

Run plesty-server <argv> in-process against this bench; capture everything.

Parameters:

argv (collections.abc.Sequence[Any])

Return type:

dict[str, Any]

fetch(path: str) → dict[str, Any]

Read a file under the bench home; path is relative to it.

Parameters:

path (str)

Return type:

dict[str, Any]

serve(stop: threading.Event, poll_ms: int = 100, max_replies: int | None = None) → int

Bind and answer requests until stop is set; returns the number answered.

Parameters:

  • stop (threading.Event) – Set to end the loop.

  • poll_ms (int) – Socket poll interval.

  • max_replies (int | None) – Stop once this many replies went out (tests).

Return type:

int

plesty.server.services.serve_in_thread(agent: Agent, stop: threading.Event | None = None) → tuple[threading.Thread, threading.Event]

Run agent on a daemon thread; returns the thread and its stop event.

Parameters:

  • agent (Agent)

  • stop (threading.Event | None)

Return type:

tuple[threading.Thread, threading.Event]

class plesty.server.services.Catalogue(index_url: str | None = None, fetch: Fetcher | None = None)

The hub catalogue at one index URL.

Bind to an index.

Parameters:

  • index_url (str | None) – /api/modules/ endpoint; the public hub by default.

  • fetch (Fetcher | None) – Reads a URL; injected by tests.

index_url = 'https://hub.plesty.net/api/modules/'
_fetch
_cache: list[plesty.server.model.catalogue.CatalogueEntry] | None = None
entries(refresh: bool = False) → list[plesty.server.model.catalogue.CatalogueEntry]

Every module the index lists (fetched once, then cached).

Raises:

CatalogueError – When the index cannot be fetched or is not JSON.

Parameters:

refresh (bool)

Return type:

list[plesty.server.model.catalogue.CatalogueEntry]

search(query: str = '', vendor: str = '', category: str = 'devices', standard: str = REQUIRED_STANDARD) → list[plesty.server.model.catalogue.CatalogueEntry]

Entries matching query, filtered by vendor, category and standard ("" = any).

Parameters:

  • query (str)

  • vendor (str)

  • category (str)

  • standard (str)

Return type:

list[plesty.server.model.catalogue.CatalogueEntry]

vendors(category: str = 'devices') → list[str]

The vendor groups present in category.

Parameters:

category (str)

Return type:

list[str]

lookup(package: str) → plesty.server.model.catalogue.CatalogueEntry | None

The entry named package (any category, any standard), or None.

Parameters:

package (str)

Return type:

plesty.server.model.catalogue.CatalogueEntry | None

exception plesty.server.services.CatalogueError

Bases: RuntimeError

The catalogue could not be fetched or read.

Initialize self. See help(type(self)) for accurate signature.

plesty.server.services.http_fetch(url: str, timeout: float = 15.0) → bytes

GET url with the standard library (the bench may have no requests).

Parameters:

  • url (str)

  • timeout (float)

Return type:

bytes

plesty.server.services.pypi_release(package: str, fetch: Fetcher | None = None) → str | None

The newest release of package on PyPI, or None when it has none there.

Parameters:

  • package (str) – Distribution name.

  • fetch (Fetcher | None) – Reads a URL; injected by tests.

Return type:

str | None

class plesty.server.services.JobRunner(home: plesty.server.model.home.Home, store: plesty.server.model.job.JobStore | None = None)

Start, observe and cancel jobs under a Home.

Bind to a home.

Parameters:
home
store
exit_file(job_id: str) → pathlib.Path

Where the launcher records the exit code of job_id.

Parameters:

job_id (str)

Return type:

pathlib.Path

start(name: str, kind: str, command: list[str], cwd: str | os.PathLike[str], env: collections.abc.Mapping[str, str] | None = None, detail: collections.abc.Mapping[str, Any] | None = None) → plesty.server.model.job.Job

Launch command as a job and return its record.

Parameters:

  • name (str) – Device instance the job belongs to.

  • kind (str) – Job kind, part of the id.

  • command (list[str]) – The command line.

  • cwd (str | os.PathLike[str]) – Working directory.

  • env (collections.abc.Mapping[str, str] | None) – Environment on top of the bench’s own.

  • detail (collections.abc.Mapping[str, Any] | None) – Kind-specific data stored on the record.

Raises:

JobError – When the command cannot be launched.

Return type:

plesty.server.model.job.Job

get(job_id: str) → plesty.server.model.job.Job | None

The job job_id with its state brought up to date, or None.

Parameters:

job_id (str)

Return type:

plesty.server.model.job.Job | None

require(job_id: str) → plesty.server.model.job.Job

Like get() but raises JobError when unknown.

Parameters:

job_id (str)

Return type:

plesty.server.model.job.Job

list(name: str | None = None, kind: str | None = None) → list[plesty.server.model.job.Job]

Jobs on record, newest first, brought up to date.

Parameters:

  • name (str | None)

  • kind (str | None)

Return type:

list[plesty.server.model.job.Job]

wait(job_id: str, timeout: float | None = None, poll: float = 0.25) → plesty.server.model.job.Job

Block until the job finishes (or timeout seconds pass); returns it.

Parameters:

  • job_id (str)

  • timeout (float | None)

  • poll (float)

Return type:

plesty.server.model.job.Job

cancel(job_id: str) → bool

Kill a running job (its whole process tree); True when it was running.

Parameters:

job_id (str)

Return type:

bool

plesty.server.services.PROBE_HOST = '127.0.0.1'
exception plesty.server.services.AlreadyRunning

Bases: SupervisorError

start was asked for a device whose server is already up.

Initialize self. See help(type(self)) for accurate signature.

exception plesty.server.services.NotRunning

Bases: SupervisorError

stop was asked for a device that has no live server.

Initialize self. See help(type(self)) for accurate signature.

exception plesty.server.services.PortBusy

Bases: SupervisorError

start found the device’s port already answering — another server holds it.

Initialize self. See help(type(self)) for accurate signature.

class plesty.server.services.Supervisor(home: plesty.server.model.home.Home, ready_timeout: float | None = None, stop_timeout: float | None = None, probe_timeout_ms: int | None = None)

Launch and control device servers under a Home.

Bind the supervisor to a home.

Parameters:

  • home (plesty.server.model.home.Home) – The bench home.

  • ready_timeout (float | None) – Seconds start() waits for the first probe answer.

  • stop_timeout (float | None) – Seconds stop() waits before escalating.

  • probe_timeout_ms (int | None) – Milliseconds one probe waits for its reply.

  • running.) ((Defaults come from config.yaml)

home
ready_timeout
stop_timeout
probe_timeout_ms
record(name: str) → plesty.server.model.process.ProcessRecord | None

The persisted launch record of name, or None.

Parameters:

name (str)

Return type:

plesty.server.model.process.ProcessRecord | None

exit_file(name: str) → pathlib.Path

Where the launcher records the exit code of the server of name.

Parameters:

name (str)

Return type:

pathlib.Path

known() → list[str]

Names that have a launch record (alive or not).

Return type:

list[str]

_drop_record(name: str) → None
Parameters:

name (str)

Return type:

None

alive(name: str) → bool

Whether the recorded server of name still runs.

Parameters:

name (str)

Return type:

bool

poll(name: str) → int | None

Exit code of the server of name, None while it runs.

LOST_EXIT when the process is gone without a recorded exit (the bench rebooted, or someone killed the launcher) — and when nothing is known at all.

Parameters:

name (str)

Return type:

int | None

status(name: str, probe_port: bool = True) → plesty.server.model.process.ProcessStatus

Observe name: record, process existence and (optionally) a probe.

Parameters:

  • name (str) – Device instance name.

  • probe_port (bool) – Whether to send a describe to tell running from starting; skipping it is cheaper for bulk listings.

Return type:

plesty.server.model.process.ProcessStatus

wait_ready(name: str, timeout: float | None = None) → bool

Poll until the server of name answers a probe.

Returns:

True once it answers; False on timeout or when the process exits first.

Parameters:

  • name (str)

  • timeout (float | None)

Return type:

bool

start(spec: plesty.server.model.device.DeviceSpec, python: pathlib.Path, wait: bool = True, timeout: float | None = None, package: str = '', version: str = '') → plesty.server.model.process.ProcessStatus

Launch the server of spec with the interpreter python.

Parameters:

  • spec (plesty.server.model.device.DeviceSpec) – Module, arguments, environment and port to run with.

  • python (pathlib.Path) – Interpreter of the environment the module is installed in.

  • wait (bool) – Block until the server answers a probe (or fails).

  • timeout (float | None) – Readiness wait in seconds; defaults to ready_timeout.

  • package (str) – Distribution name, stored on the record for listings.

  • version (str) – Installed version, likewise.

Returns:

The status after launch — running when it came up, starting when wait is off or it has not answered yet, exited when it died before answering (detail carries the log tail).

Raises:
Return type:

plesty.server.model.process.ProcessStatus

_start_locked(spec: plesty.server.model.device.DeviceSpec, python: pathlib.Path, wait: bool, timeout: float | None, package: str, version: str) → plesty.server.model.process.ProcessStatus
Parameters:
Return type:

plesty.server.model.process.ProcessStatus

stop(name: str, timeout: float | None = None) → bool

Interrupt the server of name, escalating to a kill if it lingers.

Interrupt → wait timeout → terminate → wait 3 s → kill tree.

Returns:

True when a process was stopped, False when only a stale record was cleared.

Raises:

NotRunning – When nothing (not even a stale record) is known.

Parameters:

  • name (str)

  • timeout (float | None)

Return type:

bool

restart(spec: plesty.server.model.device.DeviceSpec, python: pathlib.Path, wait: bool = True, timeout: float | None = None) → plesty.server.model.process.ProcessStatus

Stop the server of spec if it runs, then start() it again.

Parameters:
Return type:

plesty.server.model.process.ProcessStatus

relaunch(record: plesty.server.model.process.ProcessRecord, wait: bool = True) → plesty.server.model.process.ProcessStatus

Start the server again exactly as record says it was started.

Used after a host-tier field test: the interpreter, arguments and environment of the interrupted launch are what the bench expects back.

Parameters:
Return type:

plesty.server.model.process.ProcessStatus

stop_all(names: list[str] | None = None) → list[str]

Stop every known (or the given) server; returns the names stopped.

Parameters:

names (list[str] | None)

Return type:

list[str]

tail(name: str, lines: int = 50) → str

The last lines lines of the log of name ("" when no log).

Parameters:

  • name (str)

  • lines (int)

Return type:

str

_wait_exit(name: str, timeout: float) → bool
Parameters:

  • name (str)

  • timeout (float)

Return type:

bool

exception plesty.server.services.SupervisorError

Bases: RuntimeError

A start or stop request could not be carried out.

Initialize self. See help(type(self)) for accurate signature.

plesty.server.services.probe(port: int, timeout_ms: int = 2000, host_addr: str = PROBE_HOST) → bool

Ask the server on port to describe itself.

Returns:

True when a plesty-lib server answered status: ok in time.

Parameters:

  • port (int)

  • timeout_ms (int)

  • host_addr (str)

Return type:

bool

plesty.server.services.request(port: int, message: dict[str, Any], timeout_ms: int = 5000, host_addr: str = PROBE_HOST) → dict[str, Any]

Send one request to the server on port and return its reply.

The same DEALER socket probe() uses, with the message left to the caller: this is how a developer asks a running device server anything its plesty-lib command solver understands, without a client library.

Raises:

SupervisorError – When nobody answers in time, or the answer is not JSON the wire format allows.

Parameters:

  • port (int)

  • message (dict[str, Any])

  • timeout_ms (int)

  • host_addr (str)

Return type:

dict[str, Any]