pyantz.infrastructure.runner._remotes package

pyantz.infrastructure.runner._remotes package#

Submodules#

pyantz.infrastructure.runner._remotes._transfers module#

Utilities for packaging and transferring files to remote execution environments.

get_setup_tar() bundles the resolved pipeline config and any extra Python requirements into an in-memory tar archive. get_cmd() builds the uvx pyantz … invocation that the remote side runs after extracting the archive.

DEFAULT_CONTAINER_WORKING_DIR = '/workspace'#: Where containers run when the user doesn’t choose a working directory.

EXCLUDED_DIRECTORIES = frozenset({'.git', '.mypy_cache', '.pytest_cache', '.ruff_cache', '.venv', '__pycache__', 'build', 'dist', 'node_modules'})#: Directories (at any depth) which are never copied with the project.

EXCLUDED_SUFFIXES = ('.db', '.db-wal', '.db-shm', '.pyc')#: Endings of files which are never copied with the project.

_exclude_unneeded(info: TarInfo)[source]#

Skip version control, environments, caches, databases and build outputs.

Used as the filter when adding the project to an archive.

Parameters:: info (TarInfo) – an entry about to be added to the archive.
Returns:: the entry, or None to leave it out.
Return type:: TarInfo | None
Raises:: None. –

Example

import tarfile

from pyantz.infrastructure.runner._remotes._transfers import (
    _exclude_unneeded,
)

for name in ("pyantz/src/a.py", "pyantz/.git/HEAD", "pyantz/queue.db"):
    print(name, _exclude_unneeded(tarfile.TarInfo(name)) is not None)

pyantz/src/a.py True
pyantz/.git/HEAD False
pyantz/queue.db False

add_to_tar_file(contents: str, name: str, tar: TarFile)[source]#

Add our in-memory contents to the tarball.

Parameters:

contents (str) – the text of the file, written as UTF-8.
name (str) – the file’s name in the archive.
tar (TarFile) – the archive, open for writing.

Return type:

None

Returns:

None.

Raises:

tarfile.TarError – if the file can’t be added.

Example

import io
import tarfile

from pyantz.infrastructure.runner._remotes._transfers import add_to_tar_file

buffer = io.BytesIO()
with tarfile.open(fileobj=buffer, mode="w") as tar:
    add_to_tar_file("hello", "greeting.txt", tar)
buffer.seek(0)
with tarfile.open(fileobj=buffer) as tar:
    print(tar.extractfile("greeting.txt").read())

b'hello'

find_project_root(start: Path)[source]#

Find the project root: the closest parent of start with a pyproject.toml.

Parameters:: start (Path) – the directory to start looking from (included).
Returns:: the closest directory holding a pyproject.toml.
Return type:: Path
Raises:: FileNotFoundError – if there is none (for example, when pyantz was installed as a package rather than run from a source checkout)

Example

from pathlib import Path

from pyantz.infrastructure.runner._remotes._transfers import (
    find_project_root,
)

Path("project/src/pkg").mkdir(parents=True)
Path("project/pyproject.toml").touch()
print(find_project_root(Path("project/src/pkg")))

project

get_cmd(host_config: ContainerConfig | SshConfig, working_dir: str, addl_requirements: list[str] | None = None)[source]#

Get the command to run on the remote.

Unless the project directory is copied (and added through addl_requirements), the remote runs the same version of pyantz as the local machine, and has working_dir as its cwd, so that the configuration means the same thing on both.

Parameters:

host_config (ContainerConfig | SshConfig) – the remote host’s configuration.
working_dir (str) – the directory on the remote holding config.json.
addl_requirements (list[str] | None) – more arguments for uvx, e.g., ["--with", path].

Returns:

the command.

Return type:

list[str]

Raises:

None. –

Example

from pyantz import __version__
from pyantz.infrastructure.config import ContainerConfig
from pyantz.infrastructure.runner._remotes._transfers import get_cmd

cmd = get_cmd(ContainerConfig(), "/workspace")
print(cmd[2] == f"pyantz=={__version__}", cmd[3:])

True ['pyantz', '/workspace/config.json']

get_environment(host_config: ContainerConfig | SshConfig)[source]#

Collect the environment variables for the remote command.

Variables from environment_variables_file (KEY=VALUE lines, read from this machine) are overridden by environment_variables. In the file, blank lines and # comments are ignored, a leading export is allowed and quotes around values are removed.

Parameters:

host_config (ContainerConfig | SshConfig) – the remote host’s configuration.

Returns:

the environment variables.

Return type:

dict[str, str]

Raises:

OSError – if the file can’t be read.
ValueError – if a line of the file isn’t KEY=VALUE.

Example

from pathlib import Path

from pyantz.infrastructure.config import ContainerConfig
from pyantz.infrastructure.runner._remotes._transfers import get_environment

Path("vars.env").write_text("# settings\nexport A='1'\nB=2\n")
host = ContainerConfig(
    environment_variables_file="vars.env",
    environment_variables={"B": "override"},
)
print(get_environment(host))

{'A': '1', 'B': 'override'}

get_project_dir()[source]#

Copy the project directory, without environments, caches and VCS files.

Parameters:: None.
Returns:: a gzipped tar archive of the project, under pyantz/.
Return type:: BytesIO
Raises:: FileNotFoundError – if pyantz isn’t run from a source checkout.

get_setup_tar(config: InitialConfig, host_config: ContainerConfig | SshConfig)[source]#

Get a byte stream of a tar-file with the config.json and requirements file.

The configuration in config.json runs on the host’s subsequent_host; requirements.txt is only added if the host has requirements.

Parameters:

config (InitialConfig) – the pipeline’s configuration.
host_config (ContainerConfig | SshConfig) – the remote host the archive is for.

Returns:

a gzipped tar archive, positioned at its start.

Return type:

BytesIO

Raises:

None. –

Example

import tarfile

from pyantz.infrastructure.config import ContainerConfig, InitialConfig
from pyantz.infrastructure.runner._remotes._transfers import get_setup_tar

host = ContainerConfig(requirements=["polars"])
config = InitialConfig(
    jobs=[],
    submitter={"type_": "local_proc", "working_directory": "."},
    host=host,
)
with tarfile.open(fileobj=get_setup_tar(config, host)) as tar:
    print(sorted(tar.getnames()))

['config.json', 'requirements.txt']

pyantz.infrastructure.runner._remotes.container module#

Execute a pyantz pipeline inside a Docker/Podman container.

run_container() creates a container from the configured image, copies the setup tarball into it, starts it, waits for completion, and optionally copies results back to the host.

CONFLICT = 409#: HTTP status of a name clash.

_ensure_image(client: DockerClient, image: str)[source]#

Pull the image if it isn’t available locally.

Parameters:

client (DockerClient) – the container service’s client.
image (str) – the image’s name.

Return type:

None

Returns:

None.

Raises:

docker.errors.APIError – if the image can’t be pulled.

_extract_output(bits: Iterable[bytes], output_dir: str)[source]#

Extract the archive of the working directory into the output directory.

The archive has every entry under the directory’s name (workspace/...); that leading component is removed so results land directly in output_dir.

Parameters:

bits (Iterable[bytes]) – the archive, in chunks, as returned by the container service.
output_dir (str) – the directory to extract into.

Return type:

None

Returns:

None.

Raises:

tarfile.TarError – if the archive is invalid or has unsafe entries.

Example

import io
import tarfile
from pathlib import Path

from pyantz.infrastructure.runner._remotes._transfers import add_to_tar_file
from pyantz.infrastructure.runner._remotes.container import _extract_output

buffer = io.BytesIO()
with tarfile.open(fileobj=buffer, mode="w") as tar:
    add_to_tar_file("done", "workspace/result.txt", tar)
_extract_output([buffer.getvalue()], "results")
print(Path("results/result.txt").read_text())

done

_make_client(service: str)[source]#

Create a client for the configured container service.

Podman is reached through its docker compatible socket: CONTAINER_HOST if set, otherwise the default rootless socket of the current user.

Parameters:: service (str) – docker or podman.
Returns:: the client.
Return type:: DockerClient
Raises:: docker.errors.DockerException – if the service can’t be reached.

run_container(config: InitialConfig)[source]#

Copy our files to the container and run!.

A container is created from the configured image (with the configured name, if any), the configuration (and, if asked, the project) is copied into it, and it runs pyantz until done. The working directory is copied to output_dir if set, and the container is removed.

Parameters:

config (InitialConfig) – the pipeline’s configuration, with a container host.

Return type:

None

Returns:

None.

Raises:

RuntimeError – if the host isn’t a container, a container with the name already exists, or the container exits with a non-zero status.
docker.errors.DockerException – if the container service fails.

pyantz.infrastructure.runner._remotes.ssh module#

Execute a pyantz pipeline on a remote host via SSH.

run_ssh() connects to the remote machine with Paramiko, copies the setup tarball via SFTP, extracts it, and runs uvx pyantz config.json — either synchronously or as a background process (nohup … &).

REMOTE_LOG_FILE = 'pyantz.log'#: Where the output of a background run is written, in the remote working directory.

_build_remote_command(host_config: SshConfig, working_dir: str, addl_requirements: list[str])[source]#

Build the shell command which runs pyantz in the remote working directory.

Every value is quoted for the shell. With run_async, the command runs in the background with its output in REMOTE_LOG_FILE.

Parameters:

host_config (SshConfig) – the ssh host’s configuration.
working_dir (str) – the directory on the remote holding config.json.
addl_requirements (list[str]) – more arguments for uvx.

Returns:

the shell command.

Return type:

str

Raises:

OSError – if the environment variables file can’t be read.
ValueError – if the environment variables file is invalid.

Example

from pyantz.infrastructure.config import SshConfig
from pyantz.infrastructure.runner._remotes.ssh import _build_remote_command

host = SshConfig(
    remote_host="hpc",
    remote_user="me",
    copy_project_dir=True,
    environment_variables={"MODE": "fast"},
)
command = _build_remote_command(host, "/tmp/run 1", [])
print(command.replace(" > ", "\n> "))

cd '/tmp/run 1' && nohup env MODE=fast uvx pyantz '/tmp/run 1/config.json'
> pyantz.log 2>&1 < /dev/null &

_connect(ssh: SSHClient, host_config: SshConfig)[source]#

Connect, trying keys first and falling back to a password.

The password is only asked for when there is a terminal.

Parameters:

ssh (SSHClient) – the client to connect.
host_config (SshConfig) – the ssh host’s configuration.

Return type:

None

Returns:

None.

Raises:

RuntimeError – if keys don’t work and there is no terminal to ask for a password.
paramiko.SSHException – if the connection fails, e.g., the host is unknown and the policy rejects it.
OSError – if the host can’t be reached.

_host_key_policy(name: str)[source]#

Get the paramiko policy for hosts which are not in the known hosts.

Parameters:: name (str) – reject, warn or auto_add.
Returns:: the policy.
Return type:: MissingHostKeyPolicy
Raises:: KeyError – if the name is unknown.

Example

from pyantz.infrastructure.runner._remotes.ssh import _host_key_policy

print(type(_host_key_policy("warn")).__name__)

WarningPolicy

_make_remote_dir(ssh: SSHClient, path: str)[source]#

Create the directory (and any missing parents) on the remote.

Unlike sftp.mkdir this doesn’t fail when the directory already exists (for example when re-running) or when parent directories are missing.

Parameters:

ssh (SSHClient) – connected client.
path (str) – the directory to create on the remote.

Return type:

None

Returns:

None.

Raises:

RuntimeError – if mkdir -p fails on the remote.

_prepare_working_dir(ssh: SSHClient, host_config: SshConfig)[source]#

Create (or pick) the directory on the remote that the pipeline runs in.

Parameters:

ssh (SSHClient) – connected client.
host_config (SshConfig) – the ssh host’s configuration; if it has no working_dir, a new temporary directory is made on the remote.

Returns:

the path of the directory on the remote.

Return type:

str

Raises:

RuntimeError – if the directory can’t be created.

_run_checked(ssh: SSHClient, cmd: str, description: str)[source]#

Run a command on the remote and raise if it fails.

Parameters:

ssh (SSHClient) – connected client
cmd (str) – shell command to run
description (str) – what the command does, used in the error message

Returns:

what the command wrote to stdout

Return type:

str

Raises:

RuntimeError – if the exit status is not zero; the message has its stderr

_run_on_connection(ssh: SSHClient, config: InitialConfig, host_config: SshConfig)[source]#

Copy our files to the connected remote and run.

Parameters:

ssh (SSHClient) – connected client.
config (InitialConfig) – the pipeline’s configuration.
host_config (SshConfig) – the ssh host’s configuration.

Return type:

None

Returns:

None.

Raises:

RuntimeError – if a step fails on the remote.
OSError – if the files can’t be copied.
subprocess.CalledProcessError – if copying the results back fails.

_sync_output(host_config: SshConfig, working_dir: str)[source]#

Copy the remote working directory to the local output directory.

Parameters:

host_config (SshConfig) – the ssh host’s configuration, with output_dir set.
working_dir (str) – the directory on the remote to copy.

Return type:

None

Returns:

None.

Raises:

RuntimeError – if output_dir isn’t set or rsync isn’t installed.
subprocess.CalledProcessError – if rsync fails.

run_ssh(config: InitialConfig)[source]#

Copy our files to a remote and run.

Connects to the host, copies the configuration (and, if asked, the project) to the remote working directory and runs pyantz there; see SshConfig for the options.

Parameters:

config (InitialConfig) – the pipeline’s configuration, with an ssh host.

Return type:

None

Returns:

None.

Raises:

RuntimeError – if the host isn’t ssh, or a step fails on the remote.
paramiko.SSHException – if the connection fails.

Module contents#

Execute code remotely.

The pipeline’s configuration is copied to another host (over ssh, or into a container) and pyantz is run there with the host’s subsequent_host.

pyantz.infrastructure.runner._remotes package

Contents

pyantz.infrastructure.runner._remotes package#

Submodules#

pyantz.infrastructure.runner._remotes._transfers module#

pyantz.infrastructure.runner._remotes.container module#

pyantz.infrastructure.runner._remotes.ssh module#

Module contents#