srunx.web package

Subpackages

• srunx.web.routers package
  • Submodules
  • srunx.web.routers.config module
  • srunx.web.routers.files module
  • srunx.web.routers.history module
  • srunx.web.routers.jobs module
  • srunx.web.routers.resources module
  • srunx.web.routers.workflows module
  • Module contents

Submodules

srunx.web.app module

srunx.web.config module

class srunx.web.config.WebConfig(**data)

Bases: BaseModel

Web server configuration.

host: str

port: int

cors_origins: list[str]

ssh_profile: str | None

ssh_hostname: str | None

ssh_username: str | None

ssh_key_filename: str | None

ssh_port: int

model_config: ClassVar[ConfigDict] = {}

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

srunx.web.config.get_web_config()

Return type:

WebConfig

srunx.web.deps module

FastAPI dependency injection providers.

srunx.web.deps.set_adapter(adapter)

Return type:

None

srunx.web.deps.get_adapter()

Return type:

SlurmSSHAdapter

srunx.web.deps.get_history_db()

Return type:

JobHistory

srunx.web.serializers module

Serialize data for frontend-compatible responses.

In the SSH adapter architecture, most data is already dicts from the adapter’s parsing. These helpers normalize edge cases.

srunx.web.serializers.serialize_history_entry(entry)

Map history dict keys to frontend JobHistoryEntry schema.

Return type:

dict[str, Any]

srunx.web.serializers.serialize_job_stats(stats)

Map history stats dict to frontend JobStats schema.

Return type:

dict[str, Any]

srunx.web.ssh_adapter module

SSH-based SLURM adapter for the Web UI.

Wraps SSHSlurmClient to provide all operations needed by the REST API, including list_jobs, cancel_job, and get_resources which SSHSlurmClient does not natively support.

class srunx.web.ssh_adapter.SlurmSSHAdapter(*, profile_name=None, hostname=None, username=None, key_filename=None, port=22)

Bases: object

Adapter providing a unified API for the Web UI over SSH.

__init__(*, profile_name=None, hostname=None, username=None, key_filename=None, port=22)

connect()

Return type:

bool

disconnect()

Return type:

None

list_jobs(user=None)

List SLURM jobs via squeue.

Return type:

list[dict[str, Any]]

get_job(job_id)

Get detailed job info via sacct.

Return type:

dict[str, Any]

cancel_job(job_id)

Cancel a SLURM job via scancel.

Return type:

None

submit_job(script_content, job_name=None, dependency=None)

Submit a job via sbatch. Returns job info dict.

Return type:

dict[str, Any]

get_job_output(job_id, job_name=None, stdout_offset=0, stderr_offset=0)

Get job stdout/stderr log contents from remote.

Returns (stdout, stderr, new_stdout_offset, new_stderr_offset).

Return type:

tuple[str, str, int, int]

get_job_status(job_id)

Get job status string.

Return type:

str

get_resources(partition=None)

Get cluster resource information via sinfo + squeue.

Return type:

list[dict[str, Any]]

srunx.web.state module

In-memory workflow run registry.

class srunx.web.state.WorkflowRun(**data)

Bases: BaseModel

Tracks a single workflow execution.

id: str

workflow_name: str

started_at: str

completed_at: str | None

status: RunStatus

job_statuses: dict[str, str]

job_ids: dict[str, str]

error: str | None

model_config: ClassVar[ConfigDict] = {}

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

class srunx.web.state.RunRegistry

Bases: object

Thread-safe in-memory registry for workflow runs.

__init__()

create(workflow_name)

Return type:

WorkflowRun

get(run_id)

Return type:

WorkflowRun | None

list_runs(workflow_name=None)

Return type:

list[WorkflowRun]

update_job_status(run_id, job_name, status)

Update a single job’s status within a run.

Return type:

None

set_job_ids(run_id, job_ids)

Set the SLURM job ID mapping for a run.

Return type:

None

update_status(run_id, status)

Update the overall run status.

Return type:

None

complete_run(run_id, status='completed')

Mark a run as completed/failed with timestamp.

Return type:

None

fail_run(run_id, error)

Mark a run as failed with an error message.

Return type:

None

srunx.web.sync_utils module

Shared rsync sync utilities for web routers.

srunx.web.sync_utils.get_current_profile()

Get the current SSH profile from web config or ConfigManager.

Checks SRUNX_SSH_PROFILE (via get_web_config()) first, then falls back to ConfigManager.get_current_profile_name().

Returns None if no profile is configured.

Return type:

ServerProfile | None

srunx.web.sync_utils.build_rsync_client(profile)

Create RsyncClient from SSH profile, handling ssh_host vs hostname.

When ssh_host is set the client delegates all connection parameters (user, key, proxy, port) to ~/.ssh/config.

Return type:

RsyncClient

srunx.web.sync_utils.sync_mount_by_name(profile, mount_name)

Sync a named mount’s local directory to remote via rsync.

Raises:

• ValueError – If mount_name does not exist in the profile.

• RuntimeError – If the rsync process exits with a non-zero code.

Return type:

None

srunx.web.sync_utils.resolve_mounts_for_workflow(profile, jobs_data, default_project=None)

Identify mount names to sync for a workflow’s jobs.

Matches each job’s work_dir against mount remote paths using longest-prefix matching. Also includes default_project if it corresponds to a valid mount.

Return type:

list[str]

Returns:

Deduplicated list of mount names.

Module contents