jobflow_remote.config.helper module#

class jobflow_remote.config.helper.ConflictIssue(kind: str, message: str, projects: list[str] = <factory>)[source]#

Bases: object

A single configuration conflict detected across projects.

kind#

Short identifier for the type of conflict (e.g. jobs_handle_dir, queue_collection, directory). Useful for grouping when rendering.

Type:

str

message#

Human-readable description of the conflict.

Type:

str

projects#

Names of the projects involved in the conflict.

Type:

list[str]

kind: str#
message: str#
projects: list[str]#
jobflow_remote.config.helper.check_jobstore(jobstore: JobStore) → str | None[source]#
jobflow_remote.config.helper.check_projects_conflicts(projects: dict[str, Project]) → list[ConflictIssue][source]#

Detect configuration conflicts between different projects.

The rules checked are:

  • batch.jobs_handle_dir of batch workers must be unique across all batch workers that share a host. Hosts are compared via the __eq__ of the BaseHost returned by worker.get_host().

  • Queue collections (queue.store collection plus flows_collection, auxiliary_collection and batches_collection) must not be shared across projects. Equality is determined by the maggma Store __eq__.

  • Project directories (base_dir, tmp_dir, log_dir, daemon_dir) must not be shared between projects.

Parameters:

projects – Mapping of project name to Project instance, typically taken from ConfigManager.projects.

Returns:

One entry per detected conflict. Empty when no conflicts are found.

Return type:

list of ConflictIssue

jobflow_remote.config.helper.check_queue_store(queue_store: Store) → str | None[source]#
jobflow_remote.config.helper.check_worker(worker: WorkerBase, full_check: bool = False) → tuple[str | None, str | None][source]#

Check that a connection to the configured worker can be made.

jobflow_remote.config.helper.generate_dummy_exec_config() → ExecutionConfig[source]#
jobflow_remote.config.helper.generate_dummy_jobstore() → dict[source]#
jobflow_remote.config.helper.generate_dummy_project(name: str, full: bool = False) → Project[source]#
jobflow_remote.config.helper.generate_dummy_queue() → dict[source]#
jobflow_remote.config.helper.generate_dummy_worker(scheduler_type: str = 'slurm', host_type: str = 'remote') → WorkerBase[source]#