arc

ARC contrib functionality.

Class ARCWorkflow

class ARCWorkflow(*args, **kwargs)

Bases: BaseRemoteWorkflow

workflow_proxy_cls

alias of ARCWorkflowProxy

arc_workflow_run_context() → Generator[None, None, None]

Hook to provide a context manager in which the workflow run implementation is placed. This can be helpful in situations where resurces should be acquired before and released after running a workflow.

arc_job_resources(job_num: int, branches: list[int]) → dict[str, int]

Hook to define resources for a specific job with number job_num, processing branches. This method should return a dictionary.

arc_dump_intermediate_job_data() → bool

Whether to dump intermediate job data to the job submission file while jobs are being submitted.

arc_post_submit_delay() → int | float

Configurable delay in seconds to wait after submitting jobs and before starting the status polling.

arc_poll_callback(poll_data: PollData) → None

Configurable callback that is called after each job status query and before potential resubmission. It receives the variable polling attributes poll_data (PollData) that can be changed within this method. If False is returned, the polling loop is gracefully terminated. Returning any other value does not have any effect.

arc_post_poll_callback(success: bool, duration: float | int) → None

Configurable callback that is called after the polling loop has ended. It receives a boolean success that indicates whether the job polling was successful, and the duration of the job polling in seconds.

Class ARCJobManager

class ARCJobManager(job_list: str | None = None, ce: str | None = None, threads: int = 1)

Bases: BaseJobManager

submit(job_file: str | Path | Sequence[str | Path], job_list: str | None = None, ce: str | None = None, retries: int = 0, retry_delay: float | int = 3, silent: bool = False, _processes: list | None = None) → str | list[str] | None

Abstract atomic or group job submission. Can throw exceptions. Should return a single job id or a list of ids.

cancel(job_id: str | Sequence[str], job_list: str | None = None, silent: bool = False, _processes: list | None = None) → dict[str, None] | None

Abstract atomic or group job cancellation. Can throw exceptions. Should return a dictionary mapping job ids to per-job return values.

cleanup(job_id: str | Sequence[str], job_list: str | None = None, silent: bool = False, _processes: list | None = None) → dict[str, None] | None

Abstract atomic or group job cleanup. Can throw exceptions. Should return a dictionary mapping job ids to per-job return values.

query(job_id: str | Sequence[str], job_list: str | None = None, silent: bool = False, _processes: list | None = None) → dict[int, dict[str, Any]] | dict[str, Any] | None

Abstract atomic or group job status query. Can throw exceptions. Should return a dictionary mapping job ids to per-job return values.

Class ARCJobFileFactory

class ARCJobFileFactory(file_name: str = 'arc_job.xrsl', command: str | Sequence[str] | None = None, executable: str | None = None, arguments: str | Sequence[str] | None = None, input_files: dict[str, str | Path | JobInputFile] | None = None, output_files: list[str] | None = None, postfix_output_files: bool = True, output_uri: str | None = None, overwrite_output_files: bool = True, job_name: str | None = None, log: str = 'log.txt', stdout: str = 'stdout.txt', stderr: str = 'stderr.txt', custom_content: str | Sequence[str] | None = None, absolute_paths: bool = True, **kwargs)

Bases: BaseJobFileFactory

create(postfix: str | None = None, **kwargs) → tuple[str, Config]

Abstract job file creation method that must be implemented by inheriting classes.