pyantz.infrastructure.runner package

pyantz.infrastructure.runner package#

Subpackages#

Submodules#

pyantz.infrastructure.runner.job_manager module#

Run jobs.

The runners call run_job_with_details() (or run_job()) for each job they take from their queue. It fills in the job’s variables, makes them available to the job through JobVariables, and gives the job a submit function which passes on its variables and its id to the jobs it submits.

class JobResult(success: bool, details: str | None = None)[source]#

Bases: NamedTuple

Outcome of running a job.

Parameters:

  • success (bool)

  • details (str | None)

_asdict()#

Return a new dict which maps field names to their values.

_field_defaults = {'details': None}#
_fields = ('success', 'details')#
classmethod _make(iterable)#

Make a new JobResult object from a sequence or iterable

_replace(**kwds)#

Return a new JobResult object replacing specified fields with new values

details: str | None#

For failures, why (e.g., the exception raised).

success: bool#

Did the job succeed?

JobVariables = <ContextVar name='JobVariables' default=None>#

The variables of the job which is running; jobs which run other jobs (like local_for_loop()) read it to pass them on.

type SubmissionFnWithParentType = Callable[[JobWithContext, str | UUID | None], None]#

Function a runner provides to submit a job along with the id of its parent.

_run_job_no_parent_wrapper_with_details(job_config: JobWithContext, wrapped_submit_fn: SubmissionFnType)[source]#

Run the provided job and resolve variables in its parameters.

A job function reports failure by returning False or by raising an exception. Returning None (as plain python functions often do) counts as success, and any other non-bool result is judged by its truthiness; both log a warning.

Parameters:

  • job_config (JobWithContext) – the job to run, with its variables.

  • wrapped_submit_fn (SubmissionFnType) – function to submit the job’s jobs with.

Returns:

whether the job succeeded and, if not, why.

Return type:

JobResult

Raises:

ValueError – if resolving the variables finds a recursive reference.

run_job(job_config: JobWithContext, submitter_fn: SubmissionFnWithParentType)[source]#

Run the provided job and resolve variables in its parameters.

Parameters:
Returns:

True if the job succeeded.

Return type:

bool

Raises:

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

Example

from pyantz.infrastructure.config import JobWithContext
from pyantz.infrastructure.runner.job_manager import run_job

job = JobWithContext(
    function="pyantz.jobs.branching.if_else.conditionally_execute",
    parameters={
        "left_side": "%{count}",
        "comparator": ">",
        "right_side": 2,
        "if_true": {
            "function": "pyantz.jobs.testing.nop.do_nothing",
            "parameters": {},
        },
    },
    variables={"count": 3},
)
submitted = []
print(run_job(job, lambda child, parent: submitted.append(child)))
print(submitted[0].variables)

True
{'count': 3}
run_job_no_parent_wrapper(job_config: JobWithContext, wrapped_submit_fn: SubmissionFnType)[source]#

Run the provided job and resolve variables in its parameters.

Unlike run_job(), the jobs submitted by the job are passed to wrapped_submit_fn as they are, without this job’s variables or id; used by jobs which run other jobs within themselves.

Parameters:

  • job_config (JobWithContext) – the job to run, with its variables.

  • wrapped_submit_fn (SubmissionFnType) – function to submit the job’s jobs with.

Returns:

True if the job succeeded.

Return type:

bool

Raises:

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

Example

from pyantz.infrastructure.config import JobWithContext
from pyantz.infrastructure.runner.job_manager import (
    run_job_no_parent_wrapper,
)

job = JobWithContext(
    function="pyantz.jobs.testing.nop.do_nothing", parameters={}
)
print(run_job_no_parent_wrapper(job, lambda child: None))

did nothing :)
True
run_job_with_details(job_config: JobWithContext, submitter_fn: SubmissionFnWithParentType)[source]#

Run the provided job, resolve its variables, and say why it failed if so.

Jobs the job submits inherit its variables (their own take precedence) and are passed to submitter_fn with the id of this job as their parent.

Parameters:
Returns:

whether the job succeeded and, if not, why.

Return type:

JobResult

Raises:

None – failures of the job are reported in the result.

Example

from pyantz.infrastructure.config import JobWithContext
from pyantz.infrastructure.runner.job_manager import run_job_with_details

job = JobWithContext(
    function="pyantz.jobs.files.simple.copy",
    parameters={"src": "no_such_file.txt", "dst": "copy.txt"},
)
result = run_job_with_details(job, lambda child, parent: None)
print(result.success, result.details.split(":")[0])

False FileNotFoundError

pyantz.infrastructure.runner.starter module#

Startup our program.

Check if we’re in the right execution environment, the point to the correct local runner.

run_local(config: InitialConfig)[source]#

Start running the various jobs.

This is a blocking operation if running locally.

Parameters:

config (InitialConfig) – the validated configuration, whose host is this machine.

Returns:

final report of the jobs for the local runner; None for slurm, whose jobs run after this returns.

Return type:

JobsReport | None

Raises:

ValueError – if the submitter type is unknown.

Example

from pyantz.infrastructure.config import InitialConfig
from pyantz.infrastructure.runner.starter import run_local

config = InitialConfig.model_validate(
    {
        "submitter": {
            "type_": "local_proc",
            "working_directory": ".",
            "use_same_proc": True,
        },
        "jobs": [
            {
                "function": "pyantz.jobs.files.check.file_exists",
                "parameters": {"file": "missing.txt"},
            },
        ],
    }
)
report = run_local(config)
print(report.error, report.all_succeeded)

1 False
start(config: InitialConfig | Mapping[str, Any])[source]#

Check how to start the program and route to the proper starter.

This is the main entry point of pyantz, also available as pyantz.start.

Parameters:

config (InitialConfig | Mapping[str, Any]) – the pipeline’s configuration, as an InitialConfig or a mapping (e.g., loaded from JSON) describing one.

Returns:

final report of the jobs when the run is local and

blocking; None if the outcome is not known (e.g. remote hosts).

Return type:

JobsReport | None

Raises:

  • pydantic.ValidationError – if the configuration is invalid.

  • ValueError – if the submitter type is unknown.

Example

from pyantz import start

report = start(
    {
        "submitter": {"type_": "local_proc", "working_directory": "."},
        "jobs": [
            {
                "job_id": "make",
                "function": "pyantz.jobs.files.simple.create",
                "parameters": {"path": "out.txt", "contents": "hi"},
            },
            {
                "function": "pyantz.jobs.files.check.file_contains",
                "parameters": {"file": "out.txt", "substring": "hi"},
                "depends_on": ["make"],
            },
        ],
    }
)
print(report.total, report.success, report.all_succeeded)

2 2 True

Module contents#

Runners are the workers that actually perform the work of the jobs.