jobflow_remote.config package#

Submodules#

Module contents#

exception jobflow_remote.config.ConfigError[source]#

Bases: Exception

A generic Exception related to the configuration.

class jobflow_remote.config.ConfigManager(exclude_unset: bool = False, exclude_none: bool = False, warn: bool = False)[source]#

Bases: object

A manager for the projects configuration files.

Provides tool to parse project information from the selected projects folder as well as methods to update the properties of each project.

Parameters:

exclude_unset – when dumping projects determine whether fields which were not explicitly set when creating the model should be excluded from the dictionary
exclude_none – when dumping projects determine whether fields which are equal to None should be excluded from the dictionary
warn – if True print warnings related to the parsing of the files in the projects folder

backup_project(project_name: str) → None[source]#

Create the backup of a project file using the naming convention $FILENAME.bak.$N

Parameters:: project_name – Name of the project to be removed.

create_project(project: Project, ext='yaml') → None[source]#

Create a new Project in the project folder by dumping the project to file.

Parameters:

project – The data of the project to be created.
ext –

The extension of the file to which the project will be dumped (yaml, json
or toml)

dump_project(project_data: ProjectData) → None[source]#

Dump the project to filepath specified in the ProjectData.

Parameters:: project_data – The project data to be dumped

get_exec_config(exec_config_name: str, project_name: str | None = None) → ExecutionConfig[source]#

Return the ExecutionConfig object based on the name.

Parameters:

exec_config_name – Name of the ExecutionConfig.
project_name – Name of the project from which the ExecutionConfig should be retrieved, or None to use the one from the settings.

Returns:

The selected ExecutionConfig

Return type:

ExecutionConfig

get_project(project_name: str | None = None) → Project[source]#

Get the Project object based from the project name.

Parameters:: project_name – The name of the project or None to use the value from the settings
Returns:: The selected Project
Return type:: Project

get_project_data(project_name: str | None = None) → ProjectData[source]#

Get the ProjectData object based from the project name.

Parameters:: project_name – The name of the project or None to use the value from the settings
Returns:: The selected ProjectData
Return type:: ProjectData

get_worker(worker_name: str, project_name: str | None = None) → WorkerBase[source]#

Return the worker object based on the name.

Parameters:

worker_name – Name of the worker to retrieve.
project_name – Name of the project from which the Worker should be retrieved, or None to use the one from the settings.

Returns:

The selected Worker.

Return type:

WorkerBase

load_projects_data() → dict[str, ProjectData][source]#

Load projects from the selected projects folder.

Returns:: Dictionary with project name as key and ProjectData as value.
Return type:: dict

project_names_from_files(*, suppress_warnings: bool = False) → tuple[list[str], list[str]][source]#

Parses all the parsable files and only checks for the “name” attribute to return a list of potential project file names.

Useful in case some projects cannot be properly parsed, but the full list needs to be returned.

Parameters:: suppress_warnings – If set to True, suppress warnings related to the parsing of the files in the projects folder.
Returns:: List of project names + List of erroneous files
Return type:: tuple

property projects: dict[str, Project]#: returns: Dictionary with project name as key and Project as value. :rtype: dict

projects_ext = ('json', 'yaml', 'toml')#

remove_exec_config(exec_config_name: str, project_name: str | None = None) → None[source]#

Remove an ExecutionConfig from the selected project.

Parameters:

exec_config_name – Name of the ExecutionConfig to be removed
project_name – Name of the project from which the ExecutionConfig should be removed, or None to use the one from the settings.

remove_project(project_name: str, remove_folders: bool = True) → None[source]#

Remove a project from the projects folder.

Parameters:

project_name – Name of the project to be removed.
remove_folders – Optionally remove the folders related to the project (e.g. tmp, log).

remove_worker(worker_name: str, project_name: str | None = None) → None[source]#

Remove a worker from the selected project.

Parameters:

worker_name – Name of the worker to be removed
project_name – Name of the project from which the Worker should be removed, or None to use the one from the settings.

select_project_name(project_name: str | None = None) → str[source]#

Determine the project name to be used based on the passed value and on the general settings.

Parameters:: project_name – The name of the project or None to use the value from the settings
Returns:: The name of the selected project.
Return type:: str

set_exec_config(exec_config_name: str, exec_config: ExecutionConfig, project_name: str | None = None, replace: bool = False) → None[source]#

Set an ExecutionConfig in the selected project. Can add a new ExecutionConfig or replace an existing one.

Parameters:

exec_config_name – Name of the ExecutionConfig to be added or replaced.
exec_config – The ExecutionConfig.
project_name –

Name of the project where the ExecutionConfig is set, or None to use
the one from the settings.
replace – Raise an exception if False and an ExecutionConfig with the chosen name already exists.

set_jobstore(jobstore: JobStore, project_name: str | None = None) → None[source]#

Set the project specific store used for jobflow.

Parameters:

jobstore – A maggma Store
project_name – Name of the project where the Store is set, or None to use the one from the settings.

set_queue_db(store: MongoStore, project_name: str | None = None) → None[source]#

Set the project specific store used for managing the queue.

Parameters:

store – A maggma Store
project_name – Name of the project where the Store is set, or None to use the one from the settings.

set_worker(name: str, worker: WorkerBase, project_name: str | None = None, replace: bool = False) → None[source]#

Set a worker in the selected project. Can add a new worker or replace an existing one.

Parameters:

name – Name of the worker to be added or replaced.
worker – Worker to be set.
project_name – Name of the project where the Worker is set, or None to use the one from the settings.
replace – Raise an exception if False and a Worker with the chosen name already exists.

update_project(config: dict, project_name: str) → None[source]#

Update the project values. The passed dict with values will be recursively merged in the current project.

Parameters:

config – Dictionary with the project values to be updated.
project_name – Name of the project to be updated

class jobflow_remote.config.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.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]#

class jobflow_remote.config.ProjectData(filepath, project, ext)[source]#

Bases: NamedTuple

Create new instance of ProjectData(filepath, project, ext)

ext: str#: Alias for field number 2

filepath: str | Path#: Alias for field number 0

project: Project#: Alias for field number 1

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.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].