jobflow_remote.config.base module#

class jobflow_remote.config.base.BatchConfig(*, jobs_handle_dir: Path, work_dir: Path, max_jobs_per_batch: int | None = None, max_wait: float | None = 60, max_time: float | None = None, parallel_jobs: int | None = None, sleep_time: float | None = None)[source]#

Bases: BaseModel

Configuration for execution of batch jobs.

Allows to execute multiple Jobs in a single process executed on the worker (e.g. SLURM job).

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

jobs_handle_dir: Path#

max_jobs_per_batch: int | None#

max_time: float | None#

max_wait: float | None#

model_config = {'extra': 'forbid'}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

parallel_jobs: int | None#

sleep_time: float | None#

work_dir: Path#

exception jobflow_remote.config.base.ConfigError[source]#

Bases: Exception

A generic Exception related to the configuration.

Bases: BaseModel

The representation of a fabric connection. Mainly used in case of nested gateways.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

connect_kwargs: dict | None#

gateway: str | ConnectionData | None#

get_connect_kwargs() → dict[source]#

Return the fully filled connect_kwargs for Fabric.

Return type:: The RemoteHost.

host: str#

key_filename: str | list[str] | None#

model_config = {}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

passphrase: str | None#

password: str | None#

port: int | None#

user: str | None#

class jobflow_remote.config.base.ExecutionConfig(*, modules: list[str] | None = None, export: dict[str, Any] | None = None, pre_run: str | None = None, post_run: str | None = None)[source]#

Bases: BaseModel

Configuration to be set before and after the execution of a Job.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

export: dict[str, Any] | None#

model_config = {'extra': 'forbid'}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

modules: list[str] | None#

post_run: str | None#

pre_run: str | None#

class jobflow_remote.config.base.LocalWorker(*, type: Literal['local'] = 'local', scheduler_type: str | dict, work_dir: Path, resources: dict | None = None, pre_run: str | None = None, post_run: str | None = None, execution_cmd: str | None = None, timeout_execute: int = 60, max_jobs: Annotated[int | None, Ge(ge=0)] = None, batch: BatchConfig | None = None, scheduler_username: str | None = None, sanitize_command: bool = False, delay_download: int | None = None)[source]#

Bases: WorkerBase

Worker representing the local host.

Executes command directly.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

property cli_info: dict#

Short information about the worker to be displayed in the command line interface.

Return type:: A dictionary with the Worker short information.

get_host() → BaseHost[source]#

Return the LocalHost.

Return type:: The LocalHost.

model_config = {'extra': 'forbid'}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

type: Literal['local']#

class jobflow_remote.config.base.LogLevel(value)[source]#

Bases: str, Enum

Enumeration of logging level.

DEBUG = 'debug'#

ERROR = 'error'#

INFO = 'info'#

WARN = 'warn'#

to_logging() → int[source]#

Helper converter to python logging values.

Return type:: The int corresponding to python logging value

class jobflow_remote.config.base.Project(*, name: str, base_dir: str | None = None, tmp_dir: str | None = None, log_dir: str | None = None, daemon_dir: str | None = None, log_level: LogLevel = LogLevel.INFO, runner: RunnerOptions = <factory>, workers: dict[str, ~types.Annotated[~jobflow_remote.config.base.LocalWorker | ~jobflow_remote.config.base.RemoteWorker | ~jobflow_remote.config.base.SeparatedTransferWorker, FieldInfo(annotation=NoneType, required=True, discriminator='type')]] = <factory>, queue: QueueConfig, exec_config: dict[str, ~jobflow_remote.config.base.ExecutionConfig]=<factory>, jobstore: dict = <factory>, remote_jobstore: dict | None = None, metadata: dict | None = None, optional_jobstores: dict[str, dict] | None=<factory>)[source]#

Bases: BaseModel

The configurations of a Project.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

base_dir: str | None#

classmethod check_base_dir(base_dir: str, info: ValidationInfo) → str[source]#: Validator to set the default of base_dir based on the project name.

classmethod check_daemon_dir(daemon_dir: str, info: ValidationInfo) → str[source]#: Validator to set the default of daemon_dir based on the base_dir.

classmethod check_jobstore(jobstore: dict) → dict[source]#: Check that the jobstore configuration could be converted to a JobStore.

classmethod check_log_dir(log_dir: str, info: ValidationInfo) → str[source]#: Validator to set the default of log_dir based on the base_dir.

classmethod check_optional_jobstore(optional_jobstores: dict) → dict[source]#: Check that the jobstore configuration could be converted to a JobStore.

classmethod check_tmp_dir(tmp_dir: str, info: ValidationInfo) → str[source]#: Validator to set the default of tmp_dir based on the base_dir.

check_unique_jobs_handle_dir() → Project[source]#

Ensure batch.jobs_handle_dir is unique among batch workers that share a host.

Sharing the same directory on the same host among multiple batch workers leads to unpredictable behaviour at run time, since the runner uses it to exchange information with the jobs being executed. The same path on different hosts is fine, as the two filesystems are independent. Hosts are compared via the __eq__ of the BaseHost returned by worker.get_host().

daemon_dir: str | None#

exec_config: dict[str, ExecutionConfig]#

get_job_controller()[source]#

get_jobstore(name: str | None = None) → JobStore | None[source]#

Generate an instance of the JobStore based on the configuration.

Parameters:: name – name of the JobStore to fetch. If None the default JobStore, otherwise one of the optional_jobstores.
Return type:: A JobStore

get_queue_store()[source]#

Generate an instance of a maggma Store based on the queue configuration.

Return type:: A maggma Store

property has_interactive_workers: bool#: True if any of the workers have interactive_login set to True.

jobstore: dict#

log_dir: str | None#

log_level: LogLevel#

metadata: dict | None#

model_config = {'extra': 'forbid'}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

name: str#

optional_jobstores: dict[str, dict] | None#

queue: QueueConfig#

remote_jobstore: dict | None#

runner: RunnerOptions#

tmp_dir: str | None#

workers: dict[str, WorkerConfig]#

exception jobflow_remote.config.base.ProjectParsingError[source]#

Bases: ConfigError

Exception raised if the Project does not exist or could not be parsed.

exception jobflow_remote.config.base.ProjectUndefinedError[source]#

Bases: ConfigError

Exception raised if the Project has not been defined or could not be determined.

class jobflow_remote.config.base.QueueConfig(*, store: dict = <factory>, flows_collection: str = 'flows', auxiliary_collection: str = 'jf_auxiliary', batches_collection: str = 'batches', db_id_prefix: str | None = None)[source]#

Bases: BaseModel

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

auxiliary_collection: str#

batches_collection: str#

classmethod check_store(store: dict) → dict[source]#: Check that the queue configuration could be converted to a Store.

db_id_prefix: str | None#

flows_collection: str#

model_config = {'extra': 'forbid'}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

store: dict#

Bases: BaseRemoteWorker

Worker representing a remote host reached through an SSH connection.

Uses a Fabric Connection. Check Fabric documentation for more details on the options defining a Connection.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

property cli_info: dict#

Short information about the worker to be displayed in the command line interface.

Return type:: A dictionary with the Worker short information.

get_host() → BaseHost[source]#

Return the RemoteHost for this worker.

Returns:: The RemoteHost instance for this worker.
Return type:: RemoteHost

model_config = {'extra': 'forbid'}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

type: Literal['remote']#

class jobflow_remote.config.base.RunnerOptions(*, delay_checkout: float = 30, delay_check_run_status: float = 30, delay_advance_status: float = 30, delay_refresh_limited: float = 600, delay_update_batch: float = 60, delay_ping_db: float = 3600, lock_timeout: float | None = 86400, delete_tmp_folder: bool = True, max_step_attempts: int = 3, delta_retry: tuple[float, ...] = (30, 300, 1200))[source]#

Bases: BaseModel

Options to tune the execution of the Runner.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

delay_advance_status: float#

delay_check_run_status: float#

delay_checkout: float#

delay_ping_db: float#

delay_refresh_limited: float#

delay_update_batch: float#

delete_tmp_folder: bool#

delta_retry: tuple[float, ...]#

get_delta_retry(step_attempts: int) → float[source]#

The time to wait before retrying a failed advancement of the remote state, based on the number of attempts.

If exceeding the size of the list delta_retry, the last value is returned.

Parameters:: step_attempts – The number of attempts advancing a remote state.
Return type:: The delay in seconds.

lock_timeout: float | None#

max_step_attempts: int#

model_config = {'extra': 'forbid'}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class jobflow_remote.config.base.SeparatedTransferWorker(*, type: Literal['separated_transfer'] = 'separated_transfer', scheduler_type: str | dict, work_dir: Path, resources: dict | None = None, pre_run: str | None = None, post_run: str | None = None, execution_cmd: str | None = None, timeout_execute: int = 60, max_jobs: Annotated[int | None, Ge(ge=0)] = None, batch: BatchConfig | None = None, scheduler_username: str | None = None, sanitize_command: bool = False, delay_download: int | None = None, host: str, user: str | None = None, port: int | None = None, password: str | None = None, key_filename: str | list[str] | None = None, passphrase: str | None = None, gateway: str | ConnectionData | None = None, forward_agent: bool | None = None, connect_timeout: int | None = None, connect_kwargs: dict | None = None, inline_ssh_env: bool | None = None, keepalive: int | None = 60, shell_cmd: str | None = 'bash', login_shell: bool = True, interactive_login: bool = False, transfer: ConnectionData)[source]#

Bases: BaseRemoteWorker

Worker with separate hosts for commands and file transfers.

This is useful for HPC systems where login nodes have SFTP disabled but a dedicated data transfer node is available (e.g., LRC at LBNL).

Command execution goes through the main host connection, while file operations (put, get, mkdir, etc.) go through the transfer host.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

property cli_info: dict#

Short information about the worker to be displayed in the CLI.

Returns:: A dictionary with the Worker short information.
Return type:: dict

get_host() → BaseHost[source]#

Return a SeparatedTransferHost for this worker.

Creates a host that delegates commands to the main connection and file operations to the transfer connection.

Returns:: The host instance with separate command and transfer connections.
Return type:: SeparatedTransferHost

model_config = {'extra': 'forbid'}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

transfer: ConnectionData#

type: Literal['separated_transfer']#

class jobflow_remote.config.base.WorkerBase(*, type: str, scheduler_type: str | dict, work_dir: Path, resources: dict | None = None, pre_run: str | None = None, post_run: str | None = None, execution_cmd: str | None = None, timeout_execute: int = 60, max_jobs: Annotated[int | None, Ge(ge=0)] = None, batch: BatchConfig | None = None, scheduler_username: str | None = None, sanitize_command: bool = False, delay_download: int | None = None)[source]#

Bases: BaseModel

Base class defining the common field for the different types of Worker.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

batch: BatchConfig | None#

classmethod check_execution_cmd(v) → str | None[source]#

classmethod check_scheduler_type(scheduler_type: str | dict) → str | dict[source]#: Validator to set the default of scheduler_type.

classmethod check_work_dir(v) → Path[source]#

abstract property cli_info: dict#

Short information about the worker to be displayed in the command line interface.

Return type:: A dictionary with the Worker short information.

delay_download: int | None#

execution_cmd: str | None#

abstractmethod get_host() → BaseHost[source]#: Return the Host object used in the Worker.

get_scheduler_io() → BaseSchedulerIO[source]#

Get the BaseSchedulerIO from QToolKit depending on scheduler_type.

Return type:: The instance of the scheduler_type.

property is_batch: bool#

property is_local: bool#

max_jobs: int | None#

model_config = {'extra': 'forbid'}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

post_run: str | None#

pre_run: str | None#

resources: dict | None#

sanitize_command: bool#

scheduler_type: str | dict#

scheduler_username: str | None#

timeout_execute: int#

type: str#

work_dir: Path#