srunx.ssh.core package

Contents

srunx.ssh.core package#

Submodules#

srunx.ssh.core.client module#

class srunx.ssh.core.client.SlurmJob(job_id, name, status='UNKNOWN', output_file=None, error_file=None, script_path=None, is_local_script=False, _cleanup=False)[source]#

Bases: object

job_id: str#
name: str#
status: str = 'UNKNOWN'#
output_file: str | None = None#
error_file: str | None = None#
script_path: str | None = None#
is_local_script: bool = False#
__init__(job_id, name, status='UNKNOWN', output_file=None, error_file=None, script_path=None, is_local_script=False, _cleanup=False)#
class srunx.ssh.core.client.SSHSlurmClient(hostname, username, password=None, key_filename=None, port=22, proxy_jump=None, ssh_config_path=None, env_vars=None, verbose=False)[source]#

Bases: object

__init__(hostname, username, password=None, key_filename=None, port=22, proxy_jump=None, ssh_config_path=None, env_vars=None, verbose=False)[source]#
connect()[source]#
Return type:

bool

disconnect()[source]#
test_connection()[source]#

Test SSH connection and SLURM availability.

Returns:

  • ssh_connected: Whether SSH connection succeeded

  • slurm_available: Whether SLURM commands are available

  • hostname: Remote hostname

  • user: Remote username

  • slurm_version: SLURM version if available

  • error: Error message if connection failed

Return type:

Dictionary with test results including

execute_command(command)[source]#
Return type:

tuple[str, str, int]

upload_file(local_path, remote_path=None)[source]#

Upload a local file to the server and return the remote path

Return type:

str

cleanup_file(remote_path)[source]#

Remove a file from the server

Return type:

None

file_exists(remote_path)[source]#

Check if a file exists on the server

Return type:

bool

validate_remote_script(remote_path)[source]#

Validate a remote script file and return (is_valid, error_message)

Return type:

tuple[bool, str]

submit_sbatch_job(script_content, job_name=None, dependency=None)[source]#

Submit an sbatch job with script content

Return type:

SlurmJob | None

submit_sbatch_file(script_path, job_name=None, cleanup=True)[source]#

Submit an sbatch job from a local or remote file

Return type:

SlurmJob | None

cleanup_job_files(job)[source]#

Cleanup temporary files for a job if it was a local script

Return type:

None

sync_project(local_path=None, remote_path=None, *, delete=True, dry_run=False, exclude_patterns=None)[source]#

Sync the local project directory to the remote workspace via rsync.

Parameters:

  • local_path (str | None) – Local project root to sync. If None, uses git toplevel or cwd.

  • remote_path (str | None) – Remote destination. If None, uses the default ~/.config/srunx/workspace/{repo_name}/.

  • delete (bool) – Remove remote files not present locally (default True).

  • dry_run (bool) – Preview what would be transferred without syncing.

  • exclude_patterns (list[str] | None) – Additional exclude patterns for this sync.

Return type:

str

Returns:

The remote project path (for use with sbatch --chdir).

Raises:

RuntimeError – If rsync is not available or key-based auth is not configured.

get_job_status(job_id)[source]#

Get job status using SLURM commands

Return type:

str

get_job_output(job_id, job_name=None, stdout_offset=0, stderr_offset=0)[source]#

Get job output from SLURM log files.

First tries scontrol show job to discover the actual StdOut/StdErr paths configured for the job. Falls back to pattern-based search if scontrol doesn’t return usable paths.

When stdout_offset / stderr_offset are non-zero, only the bytes after that position are returned (tail-like incremental reads).

Return type:

tuple[str, str, int, int]

Returns:

(stdout, stderr, new_stdout_offset, new_stderr_offset)

get_job_output_detailed(job_id, job_name=None)[source]#

Get detailed job output information including found log files

Return type:

dict[str, str | list[str] | None]

monitor_job(job, poll_interval=10, timeout=None)[source]#

Monitor a job until completion

Return type:

SlurmJob

tail_log(job_id, job_name=None, follow=False, last_n=None, poll_interval=1.0)[source]#

Display job logs with optional real-time streaming via SSH.

Parameters:

  • job_id (str) – SLURM job ID

  • job_name (str | None) – Job name for better log file detection

  • follow (bool) – If True, continuously stream new log lines (like tail -f)

  • last_n (int | None) – Show only the last N lines

  • poll_interval (float) – Polling interval in seconds for follow mode

Returns:

  • success: Whether log retrieval was successful

  • log_content: Log content (empty in follow mode)

  • tail_command: Command to execute for follow mode (None in static mode)

  • status_message: Status or error message

  • log_file: Path to the primary log file

Return type:

Dictionary with log information

srunx.ssh.core.config module#

class srunx.ssh.core.config.MountConfig(**data)[source]#

Bases: BaseModel

Local-to-remote path mapping for a project directory.

name: str#
local: str#
remote: str#
exclude_patterns: list[str]#
expand_and_validate_paths()[source]#

Expand ~ in local path. Remote must be absolute.

Return type:

MountConfig

model_config: ClassVar[ConfigDict] = {}#

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class srunx.ssh.core.config.ServerProfile(**data)[source]#

Bases: BaseModel

hostname: str#
username: str#
key_filename: str#
port: int#
description: str | None#
ssh_host: str | None#
proxy_jump: str | None#
env_vars: dict[str, str] | None#
mounts: list[MountConfig]#
model_config: ClassVar[ConfigDict] = {}#

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class srunx.ssh.core.config.ConfigManager(config_path=None)[source]#

Bases: object

__init__(config_path=None)[source]#
load_config()[source]#
Return type:

None

save_config()[source]#

Save SSH profile data, preserving non-SSH keys (e.g. SrunxConfig).

Return type:

None

add_profile(name, profile)[source]#
Return type:

None

remove_profile(name)[source]#
Return type:

bool

get_profile(name)[source]#
Return type:

ServerProfile | None

list_profiles()[source]#
Return type:

dict[str, ServerProfile]

set_current_profile(name)[source]#
Return type:

bool

get_current_profile()[source]#
Return type:

ServerProfile | None

get_current_profile_name()[source]#
Return type:

str | None

update_profile(name, **kwargs)[source]#
Return type:

bool

expand_path(path)[source]#
Return type:

str

set_profile_env_var(profile_name, key, value)[source]#

Set an environment variable for a profile.

Return type:

bool

unset_profile_env_var(profile_name, key)[source]#

Unset an environment variable for a profile.

Return type:

bool

add_profile_mount(profile_name, mount)[source]#

Add a mount to a profile.

Return type:

bool

remove_profile_mount(profile_name, mount_name)[source]#

Remove a mount from a profile by name.

Return type:

bool

srunx.ssh.core.proxy_client module#

class srunx.ssh.core.proxy_client.ProxySSHClient(logger=None)[source]#

Bases: object

SSH Client with ProxyJump support using Paramiko

__init__(logger=None)[source]#
create_proxy_connection(proxy_host_config, target_host, target_port)[source]#

Create a proxy connection through a jump host

Return type:

Channel

close_proxy()[source]#

Close proxy connections

connect_through_proxy(target_host_config, proxy_host_name, ssh_config_path=None)[source]#

Connect to target host through proxy

Return type:

tuple[SSHClient, Channel]

srunx.ssh.core.proxy_client.create_proxy_aware_connection(hostname, username, key_filename, port=22, proxy_jump=None, ssh_config_path=None, logger=None)[source]#

Create SSH connection with optional ProxyJump support

Return type:

tuple[SSHClient, ProxySSHClient | None]

srunx.ssh.core.ssh_config module#

class srunx.ssh.core.ssh_config.SSHHost(**data)[source]#

Bases: BaseModel

hostname: str#
user: str#
port: int#
identity_file: str | None#
proxy_command: str | None#
proxy_jump: str | None#
forward_agent: bool | None#
classmethod effective_identity_file(v)[source]#
Return type:

str | None

model_config: ClassVar[ConfigDict] = {}#

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class srunx.ssh.core.ssh_config.SSHConfigParser(config_path=None)[source]#

Bases: object

__init__(config_path=None)[source]#
get_host(host_pattern)[source]#

Get host configuration by exact match or pattern matching

Return type:

SSHHost | None

list_hosts()[source]#

List all configured hosts

Return type:

dict[str, SSHHost]

find_identity_files(hostname)[source]#

Find all possible identity files for a host

Return type:

list[str]

srunx.ssh.core.ssh_config.get_ssh_config_host(hostname, config_path=None)[source]#

Convenience function to get host configuration

Return type:

SSHHost | None

Module contents#

Core SSH functionality for srunx.

This package provides low-level clients, configuration handling, and utilities that power the SSH features exposed by higher-level CLI modules.

Contents