Skip to main content

API Reference

This is the auto-generated API reference for the dagster-slurm library. It is generated directly from the docstrings in the Python source code.

Core​

This section covers the main components of the library.

Dagster Slurm Integration.

Run Dagster assets on Slurm clusters with support for:

  • Local dev mode (no SSH/Slurm)
  • Per-asset Slurm submission (staging)
  • Run-scoped Slurm Ray allocation (opt-in)
  • Session mode with operator fusion (production)
  • Multiple launchers (Bash, Ray, Spark—WIP)

class dagster_slurm.BashLauncher(**data)​

Bases: ComputeLauncher

Executes Python scripts via bash.

Uses the self-contained pixi environment extracted at runtime. Sources the activation script provided by pixi-pack.

  • Parameters: data (Any)

prepare_execution(payload_path, python_executable, working_dir, pipes_context, extra_env=None, allocation_context=None, activation_script=None)​

Generate bash execution plan.

  • Parameters:
    • payload_path (str) – Path to Python script on remote host
    • python_executable (str) – Python from extracted environment
    • working_dir (str) – Working directory
    • pipes_context (Dict[str, str]) – Dagster Pipes environment variables
    • extra_env (Optional[Dict[str, str]]) – Additional environment variables
    • allocation_context (Optional[Dict[str, Any]]) – Slurm allocation info (for session mode)
    • activation_script (Optional[str]) – Path to activation script (provided by pixi-pack)
  • Return type: ExecutionPlan
  • Returns: ExecutionPlan with shell script

class dagster_slurm.ComputeLauncher(**data)​

Bases: ConfigurableResource

Base class for compute launchers.

  • Parameters: data (Any)

prepare_execution(payload_path, python_executable, working_dir, pipes_context, extra_env=None, allocation_context=None, activation_script=None)​

Prepare execution plan.

  • Parameters:
    • payload_path (str) – Path to Python script on remote
    • python_executable (str) – Python interpreter path
    • working_dir (str) – Working directory
    • pipes_context (Dict[str, str]) – Dagster Pipes environment
    • extra_env (Optional[Dict[str, str]]) – Additional environment variables
    • allocation_context (Optional[Dict[str, Any]]) – Slurm allocation info (for session mode)
    • activation_script (Optional[str]) – Environment activation script
  • Return type: ExecutionPlan
  • Returns: ExecutionPlan with script and metadata

class dagster_slurm.ComputeResource(**data)​

Bases: ConfigurableResource

Unified compute resource - adapts to deployment.

This is the main facade that assets depend on. Hides complexity of local vs Slurm vs session execution.

Usage: : ```python @asset def my_asset(context: AssetExecutionContext, compute: ComputeResource): return compute.run( context=context, payload_path="script.py", launcher=RayLauncher(num_gpus_per_node=2) )


Configuration Examples:

Local mode (dev):

```python
compute = ComputeResource(mode="local")

Slurm per-asset mode (staging):

slurm = SlurmResource.from_env()
compute = ComputeResource(mode="slurm", slurm=slurm)

Slurm run-scoped Ray allocation (opt-in):

slurm = SlurmResource.from_env()
compute_ray = ComputeResource(
    mode="slurm",
    slurm=slurm,
    default_launcher=RayLauncher(num_gpus_per_node=1),
    allocation_scope=SlurmAllocationScope.RUN,
    run_allocation=SlurmRunAllocationConfig(num_nodes=2, gpus_per_node=1),
)

Slurm session mode with cluster reuse (prod):

slurm = SlurmResource.from_env()
session = SlurmSessionResource(slurm=slurm, num_nodes=10)
compute = ComputeResource(
    mode="slurm-session",
    slurm=slurm,
    session=session,
    enable_cluster_reuse=True,
    cluster_reuse_tolerance=0.2,
)

Heterogeneous job mode (optimal resource allocation):

compute = ComputeResource(
    mode="slurm-hetjob",
    slurm=slurm,
)
  • Parameters: data (Any)

allocation_scope : SlurmAllocationScope​

auto_detect_platform : bool​

cache_inject_globs : Optional[List[str]]​

cleanup_deferred_run_dir(context)​

Delete the remote run_dir for a step whose run() calls used defer_cleanup=True.

run(..., defer_cleanup=True) (see SlurmPipesClient.run/SlurmPipesClient.cleanup_deferred_run_dir) exists for a caller that intentionally invokes run() repeatedly for one Dagster step against the same session/allocation (for example a per-wave loop) – that pattern reuses the same deterministic run_dir on every call, so per-call async cleanup can race the next call’s own upload into that directory. Callers that opt into defer_cleanup must call this once they are truly done calling run() for that step/partition, most naturally from the same scope that decided to reuse one session/allocation across calls (e.g. right before that scope releases its allocation lease).

A no-op outside Slurm execution modes (nothing was ever deferred).

  • Return type: None

cleanup_on_failure : bool​

cluster_reuse_tolerance : float​

debug_mode : bool​

default_environment_name : Optional[str]​

default_extra_files : Optional[List[str]]​

default_launcher : Annotated[ComputeLauncher | PartialResource]​

default_skip_payload_upload : bool​

enable_cluster_reuse : bool​

get_pipes_client(context, launcher=None)​

Get appropriate Pipes client for this mode.

  • Parameters:
    • context (InitResourceContext) – Dagster resource context
    • launcher (Optional[ComputeLauncher]) – Override launcher (uses default if None)
  • Returns: LocalPipesClient or SlurmPipesClient

mode : ExecutionMode​

pack_on_remote : bool​

pack_platform : Optional[str]​

pre_deployed_env_path : Optional[str]​

project_setup_cmd : Optional[List[str]]​

project_setup_env : Dict[str, str]​

project_setup_input_globs : Optional[List[str]]​

register_cluster(cluster_address, framework, cpus, gpus, memory_gb)​

Register a newly created cluster for future reuse.

  • Parameters:
    • cluster_address (str) – Address of the cluster (e.g., “10.0.0.1:6379”)
    • framework (str) – “ray” or “spark”
    • cpus (int) – Total CPUs in cluster
    • gpus (int) – Total GPUs in cluster
    • memory_gb (int) – Total memory in GB

remote_pack_timeout : int​

run(context, payload_path, launcher=None, extra_slurm_opts=None, resource_requirements=None, force_env_push=None, skip_payload_upload=None, remote_payload_path=None, environment_name=None, config=None, extra_files=None, metrics_collector=None, poll_timeout=3600, slurm_metrics=None, **kwargs)​

Execute asset with optional resource overrides.

  • Parameters:
    • context – Dagster execution context
    • payload_path (str) – Path to Python script
    • launcher (Optional[ComputeLauncher]) – Override launcher for this asset
    • extra_slurm_opts (Optional[Dict[str, Any]]) – Override Slurm options (non-session mode)
      • nodes: int
      • cpus_per_task: int
      • mem: str (e.g., “32G”)
      • gpus_per_node: int
      • time_limit: str (e.g., “02:00:00”)
      • signal_before_timeout: str (e.g., “TERM@120”)
    • resource_requirements (Optional[Dict[str, Any]]) – Resource requirements for cluster reuse (session mode)
      • cpus: int
      • gpus: int
      • memory_gb: int
      • framework: str (“ray” or “spark”)
    • force_env_push (Optional[bool]) – Force repacking and uploading the environment for this run. If not provided, falls back to config.force_env_push or asset metadata.
    • skip_payload_upload (Optional[bool]) – If True, do not upload the payload script; expects the remote payload to already exist. Falls back to config.skip_payload_upload or asset metadata key ‘skip_slurm_payload_upload’.
    • remote_payload_path (Optional[str]) – Remote path to an existing payload when skipping upload. Falls back to config.remote_payload_path or asset metadata.
    • environment_name (Optional[str]) – Named environment produced by project_setup_cmd. Falls back to config.environment_name, slurm_environment_name asset metadata, then default_environment_name.
    • config (Optional[SlurmRunConfig]) – Optional SlurmRunConfig for run-time configuration via launchpad. Values from config are used as defaults, but explicit parameters take precedence.
    • extra_files (Optional[List[str]]) – List of local file paths to upload alongside the payload. Merged with default_extra_files, asset metadata, and config.extra_files.
    • metrics_collector (Optional[Callable[[SlurmMetricsContext], Mapping[str, Any]]]) – Optional callback that receives a SlurmMetricsContext after a standalone Slurm job or shared-allocation step finishes and returns custom Dagster metadata key-value pairs. Callback errors and invalid metadata are logged without failing the workload.
    • poll_timeout (int) – Maximum time in seconds to wait for the Slurm job to complete. Defaults to 3600 (1 hour).
    • slurm_metrics (Collection[SlurmMetric | str] | None) – Built-in Slurm metadata fields to attach. None enables all fields; an empty collection disables optional built-in fields.
    • **kwargs – Passed to client.run()
  • Yields: Dagster events
  • Return type: PipesClientCompletedInvocation

Examples​

# Simple execution with default resources
yield from compute.run(context, "script.py")
# Using SlurmRunConfig for launchpad-configurable options
@dg.asset
def my_asset(
    context: dg.AssetExecutionContext,
    compute: ComputeResource,
    config: SlurmRunConfig,
):
    return compute.run(
        context=context,
        payload_path="script.py",
        config=config,
    ).get_results()
# Override launcher for this asset
ray_launcher = RayLauncher(num_gpus_per_node=4)
yield from compute.run(context, "script.py", launcher=ray_launcher)
# Non-session mode: override Slurm resources
yield from compute.run(
    context,
    "script.py",
    extra_slurm_opts={"nodes": 1, "cpus_per_task": 16, "mem": "64G", "gpus_per_node": 2}
)
# Session mode: specify resource requirements for cluster reuse
yield from compute.run(
    context,
    "script.py",
    launcher=RayLauncher(num_gpus_per_node=2),
    resource_requirements={"cpus": 32, "gpus": 2, "memory_gb": 128, "framework": "ray"}
)

run_allocation : SlurmRunAllocationConfig​

run_hetjob(context, assets, launchers=None)​

Run multiple assets as a heterogeneous Slurm job.

Submit all assets together with their specific resource requirements. Only waits in queue ONCE, but each asset gets the resources it needs.

  • Parameters:
    • context – Dagster execution context
    • assets (List[Tuple[str, str, Dict[str, Any]]]) – List of (asset_key, payload_path, resource_requirements). Each resource_requirements dict may contain nodes (int, default 1), cpus_per_task (int, default 2), mem (str, default "4G"), gpus_per_node (int, default 0), and time_limit (str, default "01:00:00").
    • launchers (Optional[Dict[str, ComputeLauncher]]) – Optional dict mapping asset_key to ComputeLauncher
  • Yields: Dagster events

Example​

compute.run_hetjob(
    context,
    assets=[
        ("prep", "prep.py", {"nodes": 1, "cpus_per_task": 8, "mem": "32G"}),
        ("train", "train.py", {"nodes": 4, "cpus_per_task": 32, "mem": "128G", "gpus_per_node": 2}),
        ("infer", "infer.py", {"nodes": 8, "cpus_per_task": 16, "mem": "64G", "gpus_per_node": 1}),
    ],
    launchers={
        "train": RayLauncher(num_gpus_per_node=2),
        "infer": RayLauncher(num_gpus_per_node=1),
    }
)

session : Optional[SlurmSessionResource]​

slurm : Optional[SlurmResource]​

teardown(context)​

Teardown method called by Dagster at end of run. Ensures session resources and clusters are cleaned up.

  • Parameters: context (InitResourceContext)

teardown_after_execution(context)​

Release run-scoped allocations through Dagster’s resource lifecycle.

  • Parameters: context (InitResourceContext)
  • Return type: None

validate_configuration()​

Validate configuration - runs during Pydantic validation.

class dagster_slurm.LocalPipesClient(launcher, base_dir=None, require_pixi=True)​

Bases: PipesClient

Pipes client for local execution (dev mode). No SSH, no Slurm - just runs scripts locally via subprocess.

  • Parameters:

cleanup()​

Explicitly clean up resources.

run(, context, payload_path, python_executable=None, extra_env=None, extras=None, extra_slurm_opts=None, extra_files=None)​

Execute payload locally.

  • Parameters:
    • context (AssetExecutionContext) – Dagster execution context
    • payload_path (str) – Path to Python script to execute
    • python_executable (Optional[str]) – Python interpreter (defaults to current)
    • extra_env (Optional[Dict[str, str]]) – Additional environment variables
    • extras (Optional[Dict[str, Any]]) – Extra data to pass via Pipes
    • extra_slurm_opts (Optional[Dict[str, Any]])
    • extra_files (Optional[list[str]])
  • Yields: Dagster events (materializations, logs, etc.)
  • Return type: PipesClientCompletedInvocation

class dagster_slurm.RayLauncher(**data)​

Bases: ComputeLauncher

Ray distributed computing launcher.

Features:

  • Robust cluster startup with sentinel-based shutdown
  • Graceful cleanup on SIGTERM/SIGINT
  • Worker registration monitoring
  • Automatic head node detection
  • IPv4/IPv6 normalization

Modes:

  • Local: Single-node Ray
  • Cluster: Multi-node Ray cluster across Slurm allocation (via allocation_context)
  • Connect: Connect to existing cluster (via ray_address)
  • Parameters: data (Any)

dashboard_host : str​

dashboard_port : int​

grace_period : int​

head_startup_timeout : int​

network_interface : Optional[str]​

node_ip_address_command : Optional[str]​

num_gpus_per_node : int​

object_store_memory_gb : Optional[int]​

port_config : RayPortConfig​

port_strategy : Literal['fixed', 'hash_jobid', 'random']​

pre_start_commands : list[str]​

prepare_execution(payload_path, python_executable, working_dir, pipes_context, extra_env=None, allocation_context=None, activation_script=None)​

Generate Ray execution plan.

  • Parameters:
    • payload_path (str)
    • python_executable (str)
    • working_dir (str)
    • pipes_context (Dict[str, str])
    • extra_env (Optional[Dict[str, str]])
    • allocation_context (Optional[Dict[str, Any]])
    • activation_script (Optional[str])
  • Return type: ExecutionPlan

ray_address : Optional[str]​

ray_port : int​

ray_start_args : list[str]​

redis_password : Optional[str]​

use_head_ip : bool​

validate_fixed_ports()​

worker_cpu_bind : Optional[str]​

worker_startup_delay : int​

class dagster_slurm.RayPortConfig(**config_dict)​

Bases: Config

Port pool and fixed-port settings for a Ray cluster.

block_size : int​

dashboard_agent_grpc_port : int​

dashboard_agent_listen_port : int​

lock_dir : str​

max_worker_port : int​

metrics_export_port : int​

min_worker_port : int​

node_manager_port : int​

object_manager_port : int​

range_end : int​

range_start : int​

ray_client_server_port : int​

redis_shard_port : int​

runtime_env_agent_port : int​

validate_port_ranges()​

class dagster_slurm.SSHConnectionResource(**data)​

Bases: ConfigurableResource

SSH connection settings.

This resource configures a connection to a remote host via SSH. It supports key-based or password-based authentication, pseudo-terminal allocation (-t), and connections through a proxy jump host.

Supports three authentication modes:

  1. SSH key (recommended for automation) - set key_path
  2. Password (when keys are unavailable) - set password
  3. Inherited - set neither, and ssh-agent plus ~/.ssh/config decide

key_path and password are mutually exclusive.

Note that every option this resource emits overrides ~/.ssh/config, because OpenSSH prefers command-line options. That keeps a run reproducible across machines; see host_key_checking='inherit' and batch_mode=None to hand individual settings back to the operator’s own configuration.

Examples​

# Key-based auth
ssh = SSHConnectionResource(
    host="cluster.example.com",
    user="username",
    key_path="~/.ssh/id_rsa",
)

# With a proxy jump host
jump_box = SSHConnectionResource(
    host="jump.example.com", user="jumpuser", password="jump_password"
)
ssh_via_jump = SSHConnectionResource(
    host="private-cluster",
    user="user_on_cluster",
    key_path="~/.ssh/cluster_key",
    jump_host=jump_box
)

# With a post-login command (e.g., for VSC)
vsc_ssh = SSHConnectionResource(
    host="vmos.vsc.ac.at",
    user="dagster01",
    key_path="~/.ssh/vsc_key",
    force_tty=True,
    post_login_command="vsc5"
)
# From environment variables
ssh = SSHConnectionResource.from_env()
  • Parameters: data (Any)

auth_error_hint(stderr)​

Return an actionable hint for a non-interactive authentication failure.

BatchMode refuses every prompt, so a passphrase-protected key with no agent fails with a bare “Permission denied” instead of asking.

  • Parameters: stderr (str)
  • Return type: str

batch_mode : bool | None​

control_socket_path()​

Return the shared ControlMaster socket path for this connection.

  • Return type: str | None

property defers_to_ssh_config : bool​

Whether this connection adds nothing OpenSSH cannot work out itself.

Used to hand a jump host back to -J instead of rebuilding it as an explicit ProxyCommand, which would bypass ~/.ssh/config entirely.

extra_opts : list[str]​

force_tty : bool​

classmethod from_env(prefix='SLURM_SSH', _is_jump=False)​

Create from environment variables.

This method reads connection details from environment variables. The variable names are constructed using the provided prefix.

With the default prefix, the following variables are used:

  • SLURM_SSH_HOST - SSH hostname (required)
  • SLURM_SSH_PORT - SSH port (optional, default: 22)
  • SLURM_SSH_USER - SSH username (required)
  • SLURM_SSH_KEY - Path to SSH key (optional)
  • SLURM_SSH_PASSWORD - SSH password (optional)
  • SLURM_SSH_FORCE_TTY - Set to ‘true’ or ‘1’ to enable tty allocation (optional)
  • SLURM_SSH_POST_LOGIN_COMMAND - Post-login command string (optional)
  • SLURM_SSH_OPTS_EXTRA - Additional SSH options (optional)
  • SLURM_SSH_HOST_KEY_CHECKING - off, accept-new, or strict (optional)
  • SLURM_SSH_KNOWN_HOSTS_FILE - Known-hosts file path (optional)

For proxy jumps, use the _JUMP suffix for jump host variables (e.g., SLURM_SSH_JUMP_HOST, SLURM_SSH_JUMP_USER, etc.).

  • Parameters:
    • prefix (str) – Environment variable prefix (default: “SLURM_SSH”)
    • _is_jump (bool)
  • Return type: SSHConnectionResource
  • Returns: SSHConnectionResource instance

get_auth_opts()​

Build authentication options for this connection’s auth mode.

  • Return type: list[str]

get_batch_mode_opts()​

Build the BatchMode option, or nothing when it is inherited.

  • Return type: list[str]

get_common_ssh_opts(, server_alive_interval=None, server_alive_count_max=None)​

Build shared SSH/SCP options with caller overrides first.

  • Parameters:
    • server_alive_interval (int | None)
    • server_alive_count_max (int | None)
  • Return type: list[str]

get_host_key_opts()​

Build host-key verification options.

Anything emitted here overrides ~/.ssh/config, because OpenSSH always prefers command-line options. Only the settings this resource actually specifies are emitted, so host_key_checking='inherit' leaves the operator’s own configuration untouched.

  • Return type: list[str]

get_key_auth_opts(, batch_mode=None)​

Build key-based SSH authentication options.

  • Parameters: batch_mode (bool | None) – Force BatchMode for this command. When omitted the resource’s batch_mode setting applies, and None there emits no BatchMode option so ~/.ssh/config decides.
  • Return type: list[str]

get_multiplexing_opts(, create=True, control_path=None)​

Build ControlMaster options pointing at the shared control socket.

  • Parameters:
    • create (bool) – Allow the command to become the master when none exists. Callers that manage the master themselves pass False.
    • control_path (str | None) – Override the socket path (used by the connection pool).
  • Return type: list[str]

get_password_auth_opts(, prompts=3)​

Build password-based SSH authentication options.

  • Parameters: prompts (int)
  • Return type: list[str]

get_proxy_command_opts()​

Build a ProxyCommand whose SSH settings apply to the jump host.

  • Return type: list[str]

get_remote_target()​

Get the remote target string for SCP commands.

  • Return type: str

get_scp_base_command()​

Build base SCP command, including proxy and auth options.

  • Return type: list[str]

get_ssh_base_command()​

Build base SSH command, including proxy and auth options.

  • Return type: list[str]

host : str​

host_key_checking : Literal['off', 'accept-new', 'strict', 'inherit']​

host_key_error_hint(stderr)​

Return an actionable hint for an OpenSSH host-key failure.

  • Parameters: stderr (str)
  • Return type: str

jump_host : SSHConnectionResource | None​

key_path : str | None​

known_hosts_file : str | None​

property multiplexing_unsupported_reason : str | None​

Explain why ControlMaster is unavailable, if it is.

password : str | None​

port : int​

post_login_command : str | None​

property requires_tty : bool​

Return True when the resource explicitly requires a TTY.

property supports_multiplexing : bool​

Whether SSH ControlMaster can be used for this connection.

Password prompts cannot be answered by a multiplexed client, so interactive authentication (on the target or on the jump host) forces one-off connections.

user : str​

property uses_inherited_auth : bool​

Returns True when authentication is left entirely to OpenSSH.

Neither a key nor a password was configured, so ~/.ssh/config, ssh-agent and the default identity files decide.

property uses_key_auth : bool​

Returns True if using key-based authentication.

property uses_password_auth : bool​

Returns True if using password-based authentication.

class dagster_slurm.SlurmAllocation(slurm_job_id, nodes, working_dir, config)​

Bases: object

Represents a running Slurm allocation.

cancel(ssh_pool)​

Cancel the allocation.

  • Parameters: ssh_pool (SSHConnectionPool)

ensure_ray_cluster(, ssh_pool, launcher, activation_script, startup_timeout)​

Start one persistent Ray cluster inside the allocation and return its address.

  • Parameters:
    • ssh_pool (SSHConnectionPool)
    • launcher (Any)
    • activation_script (str)
    • startup_timeout (int)
  • Return type: str

execute(execution_plan, asset_key, run_dir, ssh_pool, step_update_callback=None, poll_callback=None, timeout=None)​

Execute plan in this allocation via srun.

  • Parameters:
    • execution_plan (ExecutionPlan)
    • asset_key (str)
    • run_dir (str)
    • ssh_pool (SSHConnectionPool)
    • step_update_callback (Optional[Callable[[SlurmStepExecutionResult], None]])
    • poll_callback (Optional[Callable[[SlurmStepExecutionResult], None]])
    • timeout (int | None)
  • Return type: SlurmStepExecutionResult

get_failed_nodes()​

Get list of failed nodes.

  • Return type: List[str]

is_healthy(ssh_pool)​

Check if allocation and nodes are healthy.

  • Parameters: ssh_pool (SSHConnectionPool)
  • Return type: bool

property ray_dashboard_url : str | None​

Dashboard URL reported by the persistent Ray head.

wait_for_step(result, , ssh_pool, step_update_callback=None, poll_callback=None, timeout=None)​

Wait for a launched allocation step, including after supervisor restart.

  • Parameters:
    • result (SlurmStepExecutionResult)
    • ssh_pool (SSHConnectionPool)
    • step_update_callback (Optional[Callable[[SlurmStepExecutionResult], None]])
    • poll_callback (Optional[Callable[[SlurmStepExecutionResult], None]])
    • timeout (int | None)
  • Return type: SlurmStepExecutionResult

class dagster_slurm.SlurmAllocationScope(*values)​

Bases: str, Enum

Controls how Slurm allocations are scoped for mode="slurm".

ASSET = 'asset'​

RUN = 'run'​

class dagster_slurm.SlurmMetric(*values)​

Bases: str, Enum

Optional built-in Slurm metadata fields.

ALLOCATED_GPUS = 'slurm_allocated_gpus'​

CPU_EFFICIENCY = 'cpu_efficiency_pct'​

ELAPSED_SECONDS = 'elapsed_seconds'​

EXIT_CODE = 'slurm_exit_code'​

GPU_ACCOUNTING_AVAILABLE = 'slurm_gpu_accounting_available'​

GPU_MEMORY_MAX = 'slurm_gpu_memory_max_mb'​

GPU_UTILIZATION_AVG = 'slurm_gpu_utilization_avg_pct'​

GPU_UTILIZATION_MAX = 'slurm_gpu_utilization_max_pct'​

GPU_UTILIZATION_MAX_NODE = 'slurm_gpu_utilization_max_node'​

MAX_MEMORY = 'max_memory_mb'​

NODE_HOURS = 'node_hours'​

REQUESTED_GPUS = 'slurm_requested_gpus'​

STATE = 'slurm_state'​

TRES_ACCOUNTING = 'slurm_tres_accounting'​

class dagster_slurm.SlurmMetricsContext(job_id, ssh_pool, slurm_resource, session_resource, default_metrics, step_id=None)​

Bases: object

Context passed to a custom post-run metrics collector.

  • Parameters:
    • job_id (int)
    • ssh_pool (SSHConnectionPool)
    • slurm_resource (SlurmResource)
    • session_resource (SlurmSessionResource | None)
    • default_metrics (SlurmJobMetrics)
    • step_id (str | None)

default_metrics : SlurmJobMetrics​

job_id : int​

session_resource : SlurmSessionResource | None​

slurm_resource : SlurmResource​

ssh_pool : SSHConnectionPool​

step_id : str | None = None​

class dagster_slurm.SlurmPipesClient(slurm_resource, launcher, session_resource=None, cleanup_on_failure=True, debug_mode=False, auto_detect_platform=True, pack_platform=None, pre_deployed_env_path=None, cache_inject_globs=None, pack_on_remote=False, remote_pack_timeout=600, project_setup_cmd=None, project_setup_env=None, project_setup_input_globs=None, run_allocation_scope=False)​

Bases: PipesClient

Pipes client for Slurm execution with real-time log streaming and cancellation support.

Features:

  • Real-time stdout/stderr streaming to Dagster logs
  • Packaging environment with pixi pack
  • Auto-reconnect message reading
  • Metrics collection
  • Graceful cancellation with Slurm job termination

Works in two modes:

  1. Standalone: Each asset = separate sbatch job
  2. Session: Multiple assets share a Slurm allocation (operator fusion)
  • Parameters:
    • slurm_resource (SlurmResource)
    • launcher (ComputeLauncher)
    • session_resource (Optional[SlurmSessionResource])
    • cleanup_on_failure (bool)
    • debug_mode (bool)
    • auto_detect_platform (bool)
    • pack_platform (Optional[str])
    • pre_deployed_env_path (Optional[str])
    • cache_inject_globs (Optional[list[str]])
    • pack_on_remote (bool)
    • remote_pack_timeout (int)
    • project_setup_cmd (Optional[list[str]])
    • project_setup_env (Optional[dict[str, str]])
    • project_setup_input_globs (Optional[list[str]])
    • run_allocation_scope (bool)

cleanup_deferred_run_dir(, context)​

Delete one step’s run_dir after a run of defer_cleanup=True calls.

Call this once, after the caller is certain it will not invoke run(..., defer_cleanup=True) again for the same Dagster step/partition/mapping (see _get_remote_run_dir – run_dir is derived purely from remote_base, the run id, and that identity, so it is safe to recompute here without any state carried over from the deferred calls). A natural place to call this is the same scope that opted into reusing one session/allocation across repeated run() calls (for example, right before that scope releases its allocation lease), once nothing else will write into this directory.

A no-op in debug mode, matching every other cleanup path on this client (debug mode always preserves remote directories for inspection).

  • Parameters: context (AssetExecutionContext)
  • Return type: None

run(, context, payload_path, extra_env=None, extras=None, use_session=False, extra_slurm_opts=None, force_env_push=None, skip_payload_upload=None, remote_payload_path=None, pack_cmd_override=None, pre_deployed_env_path_override=None, environment_name=None, extra_files=None, slurm_metrics=None, metrics_collector=None, poll_timeout=3600, defer_cleanup=False, **kwargs)​

Execute payload on Slurm cluster with real-time log streaming.

  • Parameters:
    • context (AssetExecutionContext) – Dagster execution context
    • payload_path (str) – Local path to Python script
    • launcher – Ignored (launcher is set at client construction time)
    • extra_env (Optional[Dict[str, str]]) – Additional environment variables
    • extras (Optional[Dict[str, Any]]) – Extra data to pass via Pipes
    • use_session (bool) – If True and session_resource provided, use shared allocation
    • extra_slurm_opts (Optional[Dict[str, Any]]) – Override Slurm options (non-session mode)
    • force_env_push (Optional[bool]) – If True, always repack and upload the environment even when a cached copy exists for the current lockfile/pack command.
    • skip_payload_upload (Optional[bool]) – If True, do not upload the payload script (assumes it already exists remotely).
    • remote_payload_path (Optional[str]) – Optional pre-existing remote payload path to use when skipping upload.
    • environment_name (Optional[str]) – Named environment prepared by project_setup_cmd.
    • slurm_metrics (Collection[SlurmMetric | str] | None) – Built-in Slurm metadata fields to attach. None enables all fields; an empty collection disables optional built-in fields.
    • metrics_collector (Callable[[SlurmMetricsContext], Mapping[str, Any]] | None) – Optional callback invoked after a standalone Slurm job or shared-allocation step reaches a terminal state. It receives a SlurmMetricsContext and returns additional Dagster metadata key-value pairs.
    • poll_timeout (int) – Maximum time in seconds to wait for the Slurm job to complete. Defaults to 3600 (1 hour).
    • defer_cleanup (bool) – If True, skip the async run_dir cleanup this call would otherwise trigger (on success, on a reattach outcome, and on failure) and leave it for the caller to remove later via cleanup_deferred_run_dir(). run_dir is deterministic per Dagster step/partition/mapping (see _get_remote_run_dir), not per call, so a caller that invokes run() repeatedly for one step against the same allocation/session (e.g. a per-wave loop that intentionally reuses one SlurmPipesClient-backed ComputeResource across many run() calls) would otherwise have each call’s fire-and-forget nohup rm -rf <run_dir> & race the next call’s mkdir -p/upload into that same, still-being-deleted directory – a real, unsynchronized delete-vs-recreate race, not merely wasted re-staging work. Defaults to False so every existing caller (one run() call per directory) is unaffected.
    • **kwargs – Additional arguments (ignored, for forward compatibility)
    • pack_cmd_override (Optional[list[str]])
    • pre_deployed_env_path_override (Optional[str])
    • extra_files (Optional[list[str]])
  • Yields: Dagster events
  • Return type: PipesClientCompletedInvocation

class dagster_slurm.SlurmQueueConfig(**data)​

Bases: ConfigurableResource

Default Slurm job submission parameters. These can be overridden per-asset via metadata or function arguments.

  • Parameters: data (Any)

account : Optional[str]​

cpus : int​

gpus_per_node : int​

mem : Optional[str]​

mem_per_cpu : Optional[str]​

num_nodes : int​

partition : str​

qos : Optional[str]​

reservation : Optional[str]​

signal_before_timeout : Optional[str]​

time_limit : str​

class dagster_slurm.SlurmResource(**data)​

Bases: ConfigurableResource

Complete Slurm cluster configuration. Combines SSH connection, queue defaults, and cluster-specific paths.

  • Parameters: data (Any)

classmethod from_env()​

Create from environment variables.

classmethod from_env_slurm(ssh)​

Create a SlurmResource by populating most fields from environment variables, but requires an explicit, pre-configured SSHConnectionResource to be provided.

next_status_poll_interval(current)​

Return the next status polling interval after an unchanged state.

  • Parameters: current (float)
  • Return type: float

queue : Annotated[SlurmQueueConfig | PartialResource]​

remote_base : Optional[str]​

set_auth_provider(provider)​

ssh : Annotated[SSHConnectionResource | PartialResource]​

status_poll_backoff_factor : float​

status_poll_interval_seconds : float​

status_poll_max_interval_seconds : float​

class dagster_slurm.SlurmRunAllocationConfig(**config_dict)​

Bases: Config

Configuration for a run-owned Slurm allocation.

account : Optional[str]​

cleanup_policy : Literal['after_run']​

constraint : Optional[str]​

cpus_per_task : Optional[int]​

gpus_per_node : Optional[int]​

mem : Optional[str]​

mem_per_cpu : Optional[str]​

nodelist : Optional[str]​

num_nodes : Optional[int]​

partition : Optional[str]​

qos : Optional[str]​

reservation : Optional[str]​

signal_before_timeout : Optional[str]​

time_limit : Optional[str]​

class dagster_slurm.SlurmRunConfig(**config_dict)​

Bases: Config

Per-run configuration for Slurm execution.

Use this to configure environment caching and payload upload behavior at job submission time via the Dagster launchpad.

Example usage in an asset:

@dg.asset
def my_asset(
    context: dg.AssetExecutionContext,
    compute: ComputeResource,
    config: SlurmRunConfig,
):
    return compute.run(
        context=context,
        payload_path="script.py",
        config=config,
    ).get_results()

Then in the Dagster launchpad, you can override:

  • force_env_push: True to force re-upload the environment
  • skip_payload_upload: True to skip uploading the payload script

environment_name : Optional[str]​

extra_files : Optional[List[str]]​

force_env_push : bool​

remote_payload_path : Optional[str]​

skip_payload_upload : bool​

class dagster_slurm.SlurmSessionResource(**data)​

Bases: ConfigurableResource

Slurm session resource for operator fusion.

This is a proper Dagster resource that manages the lifecycle of a Slurm allocation across multiple assets in a run.

Usage in definitions.py:

session = SlurmSessionResource(
    slurm=slurm,
    num_nodes=4,
    time_limit="04:00:00",
)
  • Parameters: data (Any)

account : Optional[str]​

constraint : Optional[str]​

cpus_per_task : Optional[int]​

enable_health_checks : bool​

enable_session : bool​

execute_in_session(execution_plan, asset_key, run_dir, step_update_callback=None, poll_callback=None, timeout=None)​

Execute workload in the shared allocation. Thread-safe for parallel asset execution.

  • Parameters:
    • execution_plan (ExecutionPlan)
    • asset_key (str)
    • run_dir (str)
    • step_update_callback (Optional[Callable[[SlurmStepExecutionResult], None]])
    • poll_callback (Optional[Callable[[SlurmStepExecutionResult], None]])
    • timeout (int | None)
  • Return type: SlurmStepExecutionResult

gpus_per_node : Optional[int]​

property logger : Any​

max_concurrent_jobs : int​

mem : Optional[str]​

mem_per_cpu : Optional[str]​

nodelist : Optional[str]​

num_nodes : int​

partition : Optional[str]​

qos : Optional[str]​

reservation : Optional[str]​

setup_for_execution(context)​

Called by Dagster when resource is initialized for a run. This is the proper Dagster resource lifecycle hook.

  • Parameters: context (InitResourceContext)
  • Return type: None

signal_before_timeout : Optional[str]​

slurm : SlurmResource​

teardown_after_execution(context)​

Called by Dagster when resource is torn down after run completion. This is the proper Dagster resource lifecycle hook.

  • Parameters: context (InitResourceContext)
  • Return type: None

time_limit : str​

class dagster_slurm.SparkLauncher(**data)​

Bases: ComputeLauncher

Apache Spark launcher.

Modes:

  • Local: Single-node Spark (no allocation_context)
  • Cluster: Spark cluster across Slurm allocation (via allocation_context)
  • Standalone: Connect to existing Spark cluster (via master_url)
  • Parameters: data (Any)

driver_memory : str​

executor_cores : int​

executor_memory : str​

master_url : Optional[str]​

num_executors : Optional[int]​

prepare_execution(payload_path, python_executable, working_dir, pipes_context, extra_env=None, allocation_context=None, activation_script=None)​

Generate Spark execution plan.

  • Parameters:
    • payload_path (str)
    • python_executable (str)
    • working_dir (str)
    • pipes_context (Dict[str, str])
    • extra_env (Optional[Dict[str, str]])
    • allocation_context (Optional[Dict[str, Any]])
    • activation_script (Optional[str])
  • Return type: ExecutionPlan

spark_home : str​

dagster_slurm.build_slurm_orphan_reconcile_sensor(slurm_resource, , name='slurm_orphan_reconcile_sensor', jobs=None, target=None, stale_after_seconds=120.0, limit=50, minimum_interval_seconds=30, default_status=DefaultSensorStatus.STOPPED)​

Build a sensor that retries orphaned Slurm-backed Dagster runs.

  • Parameters:
    • slurm_resource (SlurmResource)
    • name (str)
    • jobs (Sequence[Any] | None)
    • target (Any | None)
    • stale_after_seconds (float)
    • limit (int)
    • minimum_interval_seconds (int)
    • default_status (DefaultSensorStatus)
  • Return type: SensorDefinition

dagster_slurm.reconcile_orphaned_slurm_runs(instance, slurm_resource, , stale_after_seconds=120.0, limit=50, now=None)​

Mark dead-supervisor runs failed and return reattach retry requests.

The returned RunRequest objects carry the original Slurm job id and run directory. SlurmPipesClient consumes those tags on the retry and replays the remote messages.jsonl file instead of submitting another job.

  • Parameters:
    • instance (DagsterInstance)
    • slurm_resource (SlurmResource)
    • stale_after_seconds (float)
    • limit (int)
    • now (float | None)
  • Return type: list[RunRequest]

dagster_slurm.run_with_ray_reserve_topup(primary, reserve, minimum_resources, , stable_polls=2, poll_interval_seconds=1.0, timeout_seconds=None, resource_provider=None, provider_retry_attempts=3, provider_retry_backoff_seconds=1.0)​

Run primary work immediately and top it up after capacity stabilizes.

primary and the capacity watcher start concurrently. Once the watcher confirms persistent idle resources, reserve receives the qualifying snapshot and runs alongside any primary work still in progress. If the capacity wait times out, the primary result is returned with no reserve result. The callbacks can each create and materialize an independent Ray actor pool; callers remain responsible for merging their returned data.

  • Return type: tuple[TypeVar(PrimaryResultT), Optional[TypeVar(ReserveResultT)]]
  • Returns: A pair containing the primary result and the optional reserve result.
  • Parameters:
    • primary (Callable[[], TypeVar(PrimaryResultT)])
    • reserve (Callable[[Mapping[str, float]], TypeVar(ReserveResultT)])
    • minimum_resources (Mapping[str, float])
    • stable_polls (int)
    • poll_interval_seconds (float)
    • timeout_seconds (float | None)
    • resource_provider (Callable[[], Mapping[str, float]] | None)
    • provider_retry_attempts (int)
    • provider_retry_backoff_seconds (float)

dagster_slurm.wait_for_stable_ray_resources(minimum_resources, , stable_polls=2, poll_interval_seconds=1.0, timeout_seconds=None, resource_provider=None, provider_retry_attempts=3, provider_retry_backoff_seconds=1.0)​

Wait until Ray resources stay available across consecutive polls.

This observes capacity; it does not reserve it. Callers should submit the work that claims the returned capacity immediately. A common use is to launch a second Ray actor pool after co-tenant workloads release resources, because an actor pool that is already running cannot be resized.

  • Parameters:
    • minimum_resources (Mapping[str, float]) – Resource quantities that must all be available, such as {"GPU": 1, "CPU": 4}.
    • stable_polls (int) – Number of consecutive qualifying snapshots required.
    • poll_interval_seconds (float) – Delay between resource snapshots.
    • timeout_seconds (float | None) – Maximum wait time, or no limit when omitted.
    • resource_provider (Callable[[], Mapping[str, float]] | None) – Optional provider compatible with ray.available_resources. Primarily useful for custom Ray integrations and deterministic tests.
    • provider_retry_attempts (int) – Total attempts allowed for each resource snapshot when the provider raises.
    • provider_retry_backoff_seconds (float) – Initial retry delay. Subsequent delays use exponential backoff.
  • Return type: dict[str, float]
  • Returns: The final qualifying resource snapshot.
  • Raises:
    • ImportError – If Ray is not installed and no provider is supplied.
    • RuntimeError – If the resource provider exhausts its retry attempts.
    • TimeoutError – If capacity does not stabilize before the timeout.
    • ValueError – If polling or resource requirements are invalid.