pyantz.jobs.wrappers package

pyantz.jobs.wrappers package#

Submodules#

pyantz.jobs.wrappers.local module#

Run a sequence of jobs synchronously within the current process.

run_pipeline() executes jobs one after another, short-circuiting on the first failure. It does not submit jobs to the queue — all execution happens inline, which is more efficient for very small, fast steps.

Use this when you want to avoid queue overhead for a tight inner loop, or when you need to share in-memory state between steps without a file system.

class LocalPipelineParameters(*, jobs: JobPipeline, inter_job_named_pipes: list[str] = <factory>)[source]#

Bases: BaseModel

Parameters for running jobs sequentially within this job without submitting.

Parameters:

jobs (JobPipeline)
inter_job_named_pipes (list[str])

_abc_impl = <_abc._abc_data object>#

inter_job_named_pipes: list[str]#

accepted so that existing configurations keep validating, but the value is ignored.

Type:: Not implemented

jobs: JobPipeline#: The jobs to run, in order.

run_pipeline(params: LocalPipelineParameters)[source]#

Run the pipeline of jobs sequentially.

The jobs run in this process, one after another, and inherit the variables of this job. The pipeline stops at the first job which fails. The jobs can’t submit other jobs.

Parameters:

params (LocalPipelineParameters) – the job’s parameters, validated as LocalPipelineParameters.

Job parameters:

jobs (JobPipeline) – the jobs to run, in order.
inter_job_named_pipes (list[str]) – not implemented; ignored.

Returns:

True if every job succeeded, False as soon as one fails (including a job which tries to submit another job).

Return type:

bool

Raises:

None – failures of the jobs are reported by returning False.

Example

import os

from pyantz.jobs.wrappers.local import run_pipeline

touch = "pyantz.jobs.testing.touch_file.touch_file"
success = run_pipeline(
    {
        "jobs": [
            {"function": touch, "parameters": {"path": "first.txt"}},
            {"function": touch, "parameters": {"path": "second.txt"}},
        ],
    },
    lambda job: None,
)
print(success, os.path.exists("first.txt"), os.path.exists("second.txt"))

True True True

pyantz.jobs.wrappers.loops module#

Jobs for repeating a pipeline a fixed number of times.

local_for_loop() runs a sequence of jobs synchronously N times in the current process. All jobs in the inner pipeline share the same variable context from the enclosing scope.

class LocalForLoopParams(*, num_iterations: int, jobs: JobPipeline, index_variable: str = 'loop_index', stop_on_failure: bool = True)[source]#

Bases: BaseModel

Run a loop of jobs locally.

Parameters:

num_iterations (int)
jobs (JobPipeline)
index_variable (str)
stop_on_failure (bool)

_abc_impl = <_abc._abc_data object>#

index_variable: str#: Name of the variable holding the index of the iteration (0, 1, …) which the jobs can use, for example in a path like out_%{loop_index}.txt.

jobs: JobPipeline#: The jobs to run in each iteration, in order.

num_iterations: int#: How many times to run the jobs.

stop_on_failure: bool#: If true, the loop ends at the first failing job; if false, the rest of the iteration and all later iterations still run, and the loop fails at the end.

local_for_loop(params: LocalForLoopParams, submit_fn: SubmissionFnType)[source]#

Run a series of jobs locally for a set of iterations.

The jobs run in this process and inherit the variables of this job, plus the index of the iteration in index_variable. Jobs submitted by the inner jobs are passed on to submit_fn.

Parameters:

params (LocalForLoopParams) – the job’s parameters, validated as LocalForLoopParams.
submit_fn (SubmissionFnType) – function the inner jobs submit their jobs with.

Job parameters:

num_iterations (int) – how many times to run the jobs.
jobs (JobPipeline) – the jobs to run in each iteration, in order.
index_variable (str) – name of the variable holding the iteration’s index. Defaults to "loop_index".
stop_on_failure (bool) – if True, stop at the first failing job; otherwise keep going and fail at the end. Defaults to True.

Returns:

True if every job of every iteration succeeded, otherwise False.

Return type:

bool

Raises:

None – failures of the jobs are reported by returning False.

Example

import os

from pyantz.jobs.wrappers.loops import local_for_loop

success = local_for_loop(
    {
        "num_iterations": 3,
        "jobs": [
            {
                "function": "pyantz.jobs.testing.touch_file.touch_file",
                "parameters": {"path": "loop_%{loop_index}.txt"},
            },
        ],
    },
    lambda job: None,
)
print(success, sorted(f for f in os.listdir() if f.startswith("loop_")))

True ['loop_0.txt', 'loop_1.txt', 'loop_2.txt']

pyantz.jobs.wrappers.submit module#

Jobs that wrap a sub-pipeline in a different submitter or host.

submit() re-wraps a list of jobs under a new LocalRunnerConfig or SlurmRunnerConfig, letting users switch execution backends mid-pipeline.

new_host() additionally moves execution to a different host (SSH or container) before using the new submitter.

class HostParams(*, host_config: HostConfig, submit_config: AnyRunner, jobs: tuple[JobConfig, ...])[source]#

Bases: BaseModel

Parameters to run jobs through a new submitter on another host.

Parameters:

host_config (HostConfig)
submit_config (AnyRunner)
jobs (tuple[JobConfig, ...])

_abc_impl = <_abc._abc_data object>#

host_config: HostConfig#: Configuration of the host to run on.

jobs: tuple[JobConfig, ...]#: The jobs to run.

submit_config: AnyRunner#: Configuration of the submitter to use on the new host.

class SubmitParams(*, runner_config: AnyRunner, jobs: tuple[JobConfig, ...])[source]#

Bases: BaseModel

Parameters to submit a job through a new submitter.

Parameters:

runner_config (AnyRunner)
jobs (tuple[JobConfig, ...])

_abc_impl = <_abc._abc_data object>#

jobs: tuple[JobConfig, ...]#: The jobs to run.

runner_config: AnyRunner#: The submitter to run the jobs with.

new_host(params: HostParams)[source]#

Run a set of jobs on a new host.

Like submit(), but the submitter first moves to another host, such as a machine reached over ssh or a container.

Parameters:

params (HostParams) – the job’s parameters, validated as HostParams.

Job parameters:

host_config (HostConfig) – the host to run on (local, ssh or container).
submit_config (AnyRunner) – the submitter to use on the new host.
jobs (tuple[JobConfig, …]) – the jobs to run.

Returns:

True if every job succeeded (or the host or submitter doesn’t report on them), otherwise False.

Return type:

bool

Raises:

pydantic.ValidationError – if the jobs don’t form a valid configuration.
Exception – errors reaching the host, e.g., an ssh connection failure, are raised by the host’s runner.

Example

import os

from pyantz.jobs.wrappers.submit import new_host

success = new_host(
    {
        "host_config": {"type_": "local"},
        "submit_config": {
            "type_": "local_proc",
            "working_directory": ".",
            "use_same_proc": True,
        },
        "jobs": [
            {
                "function": "pyantz.jobs.testing.touch_file.touch_file",
                "parameters": {"path": "on_new_host.txt"},
            },
        ],
    },
    lambda job: None,
)
print(success, os.path.exists("on_new_host.txt"))

True True

submit(params: SubmitParams)[source]#

Wrap a set of jobs in a new submitter.

This allows a user to switch submitters in their pipeline, which may prove useful in some cases.

For example, the user may want to submit a set of jobs to slurm but then have the jobs run serially locally in a simple multiprocess to reduce the impact to their “priority” level from the slurm scheduler.

The jobs inherit the variables of this job. This job waits for them when the submitter reports back (the local runner does); otherwise it succeeds once they are submitted.

Parameters:

params (SubmitParams) – the job’s parameters, validated as SubmitParams.

Job parameters:

runner_config (AnyRunner) – the submitter to run the jobs with, such as a LocalRunnerConfig.
jobs (tuple[JobConfig, …]) – the jobs to run.

Returns:

True if every job succeeded (or the submitter doesn’t report on them), otherwise False.

Return type:

bool

Raises:

pydantic.ValidationError – if the jobs don’t form a valid configuration (e.g., they depend on unknown jobs).

Example

import os

from pyantz.jobs.wrappers.submit import submit

success = submit(
    {
        "runner_config": {
            "type_": "local_proc",
            "working_directory": ".",
            "use_same_proc": True,
        },
        "jobs": [
            {
                "function": "pyantz.jobs.testing.touch_file.touch_file",
                "parameters": {"path": "submitted.txt"},
            },
        ],
    },
    lambda job: None,
)
print(success, os.path.exists("submitted.txt"))

True True

pyantz.jobs.wrappers.variables module#

Wrap jobs in variable contexts.

A user may wish to declare some variables locally for a few common jobs.

class SetVariables(*, setter_job: Callable[..., Any], set_job_kwargs: Mapping[str, Any], jobs: JobPipeline)[source]#

Bases: BaseModel

Parameters for running a job which sets variables.

Parameters:

setter_job (Annotated[Callable[[...], Any], BeforeValidator(func=~pyantz.infrastructure.config.fn_utils.import_module_item_by_name, json_schema_input_type=PydanticUndefined), WithJsonSchema(json_schema={'type': 'string'}, mode=None)])
set_job_kwargs (Mapping[str, Any])
jobs (JobPipeline)

_abc_impl = <_abc._abc_data object>#

jobs: JobPipeline#: Jobs to submit with the variables set by the setter job.

set_job_kwargs: Mapping[str, Any]#: Keyword arguments to pass to the setter job.

setter_job: Callable[..., Any]#: Import path of the function which sets the variables; it is called with set_var (a function taking a name and a value) and set_job_kwargs.

class TemporaryDirectoryParams(*, jobs: JobPipeline, variable_name: str, submit_type: Literal['local', 'remote'] = 'local', shared_working_dir: Path | None = None)[source]#

Bases: BaseModel

Params for providing a set of jobs a temporary directory.

Parameters:

jobs (JobPipeline)
variable_name (str)
submit_type (Literal['local', 'remote'])
shared_working_dir (Path | None)

_abc_impl = <_abc._abc_data object>#

jobs: JobPipeline#: Jobs to run.

shared_working_dir: Path | None#

required with remote; with local, by default the system’s temporary location.

Type:: Directory in which the temporary directory is made

submit_type: Literal['local', 'remote']#: If local, the jobs run in this job, one after another; if remote, they are submitted. With remote, the user must provide a SHARED NON-TEMPORARY location in shared_working_dir in which to place the temporary directory, since it must outlive this job.

validate_submit_and_dir()[source]#

Ensure that remote mode has a shared working directory.

Parameters:: None.
Returns:: the validated parameters.
Return type:: TemporaryDirectoryParams
Raises:: ValueError – if submit_type is remote without a shared_working_dir.

variable_name: str#: Name of the variable holding the path to the temporary directory.

class VariableContextParams(*, jobs: JobPipeline, shared_variables: Mapping[str, Any])[source]#

Bases: BaseModel

Parameters for submitting jobs with variables shared between them.

Parameters:

jobs (JobPipeline)
shared_variables (Mapping[str, Any])

_abc_impl = <_abc._abc_data object>#

jobs: JobPipeline#: Jobs to submit with the new variables.

shared_variables: Mapping[str, Any]#: Variables to add to the jobs when submitted.

_run_jobs_temp_dir_local(params: TemporaryDirectoryParams)[source]#

Run the jobs locally with a temporary directory.

Parameters:: params (TemporaryDirectoryParams) – the parameters of run_jobs_with_temp_dir().
Returns:: True if every job succeeded, False at the first which fails.
Return type:: bool
Raises:: OSError – if the temporary directory can’t be created.

_run_jobs_temp_dir_remote(params: TemporaryDirectoryParams, submit_fn: SubmissionFnType)[source]#

Submit the jobs to be run after creating a working dir for them.

Submits a job creating a uniquely named directory in shared_working_dir, the jobs (depending on it, with the directory in their variables) and a job deleting the directory, which depends on all of them.

Parameters:

params (TemporaryDirectoryParams) – the parameters of run_jobs_with_temp_dir().
submit_fn (SubmissionFnType) – function to submit the jobs with.

Returns:

True once the jobs are submitted.

Return type:

bool

Raises:

ValueError – if shared_working_dir isn’t set.

Example

from pyantz.jobs.wrappers.variables import (
    TemporaryDirectoryParams,
    _run_jobs_temp_dir_remote,
)

nop = "pyantz.jobs.testing.nop.do_nothing"
params = TemporaryDirectoryParams(
    jobs=[{"function": nop, "parameters": {}}],
    variable_name="scratch",
    submit_type="remote",
    shared_working_dir="/shared",
)
submitted = []
_run_jobs_temp_dir_remote(params, submitted.append)
print([job.name for job in submitted])

['__autogenerate_mkdir__', None, '__autogenerated_clean_up_dir__']

run_jobs_in_context(params: VariableContextParams, submit_fn: SubmissionFnType)[source]#

Submit the set of jobs with the shared variables. Not really anything fancy.

When the jobs are submitted, they also inherit the variables of this job; the shared variables take precedence over those.

Parameters:

params (VariableContextParams) – the job’s parameters, validated as VariableContextParams.
submit_fn (SubmissionFnType) – function to submit the jobs with.

Job parameters:

jobs (JobPipeline) – jobs to submit.
shared_variables (Mapping[str, Any]) – variables to give every job.

Returns:

True once the jobs are submitted.

Return type:

bool

Raises:

None. –

Example

from pyantz.jobs.wrappers.variables import run_jobs_in_context

nop = "pyantz.jobs.testing.nop.do_nothing"
submitted = []
run_jobs_in_context(
    {
        "shared_variables": {"dataset": "train", "seed": 1},
        "jobs": [
            {"function": nop, "parameters": {"step": "prepare"}},
            {"function": nop, "parameters": {"step": "fit"}},
        ],
    },
    submitted.append,
)
for job in submitted:
    print(job.parameters["step"], job.variables)

prepare {'dataset': 'train', 'seed': 1}
fit {'dataset': 'train', 'seed': 1}

run_jobs_with_temp_dir(params: TemporaryDirectoryParams, submit_fn: SubmissionFnType)[source]#

Run jobs with a temporary directory.

The path of the directory is given to the jobs in the variable_name variable. With submit_type local, the jobs run in this job, one after another, stopping at the first failure, and the directory is deleted when they finish; they can’t submit other jobs. With remote, the jobs are submitted between a job which creates the directory and one which deletes it once they have all succeeded.

Parameters:

params (TemporaryDirectoryParams) – the job’s parameters, validated as TemporaryDirectoryParams.
submit_fn (SubmissionFnType) – function to submit the jobs with (remote only).

Job parameters:

jobs (JobPipeline) – jobs to run.
variable_name (str) – name of the variable holding the directory’s path.
submit_type (str) – local or remote. Defaults to local.
shared_working_dir (Path | None) – directory to make the temporary directory in; required with remote. Defaults to None.

Returns:

with local, True if every job succeeded, otherwise False; with remote, True once the jobs are submitted.

Return type:

bool

Raises:

OSError – if the temporary directory can’t be created (local).

Example

import os

from pyantz.jobs.wrappers.variables import run_jobs_with_temp_dir

success = run_jobs_with_temp_dir(
    {
        "variable_name": "scratch",
        "shared_working_dir": ".",
        "jobs": [
            {
                "function": "pyantz.jobs.files.simple.create",
                "parameters": {"path": "%{scratch}/temp.txt"},
            },
            {
                "function": "pyantz.jobs.files.check.file_exists",
                "parameters": {"file": "%{scratch}/temp.txt"},
            },
        ],
    },
    lambda job: None,
)
print(success, os.listdir())  # the directory is gone again

True []

set_variables(params: SetVariables, submit_fn: SubmissionFnType)[source]#

Set the variables for subsequent defined jobs.

The setter_job is a slightly abnormal job. One of the parameters it accepts shall be set_var, which is a closure that will set the enclosing variables. Once it returns, jobs are submitted with the variables it set.

Parameters:

params (SetVariables) – the job’s parameters, validated as SetVariables.
submit_fn (SubmissionFnType) – function to submit jobs with.

Job parameters:

setter_job (str) – import path of the function which sets the variables by calling set_var(name, value).
set_job_kwargs (Mapping[str, Any]) – more keyword arguments for the setter job.
jobs (JobPipeline) – jobs to submit with the variables.

Returns:

True once the jobs are submitted, False if the setter job raises (and nothing is submitted).

Return type:

bool

Raises:

None – failures of the setter job are logged and reported by returning False.

Example

from pyantz.jobs.wrappers.variables import set_variables

# in a configuration, the setter job is given by its import path
def find_latest(set_var, prefix):
    set_var("latest", f"{prefix}_042")

submitted = []
set_variables(
    {
        "setter_job": find_latest,
        "set_job_kwargs": {"prefix": "run"},
        "jobs": [
            {
                "function": "pyantz.jobs.testing.nop.do_nothing",
                "parameters": {"path": "%{latest}/out.txt"},
            },
        ],
    },
    submitted.append,
)
print(submitted[0].variables)

{'latest': 'run_042'}

Module contents#

Jobs which wrap other PyAntz jobs.

pyantz.jobs.wrappers package

Contents

pyantz.jobs.wrappers package#

Submodules#

pyantz.jobs.wrappers.local module#

pyantz.jobs.wrappers.loops module#

pyantz.jobs.wrappers.submit module#

pyantz.jobs.wrappers.variables module#

Module contents#