pyantz.infrastructure.virtual package

Submodules

pyantz.infrastructure.virtual.compile module

Compile virtual jobs to ‘real’ jobs.

_check_unique_ids(jobs: list[AnyJobConfig])

Raise a ValueError if any job id is used by more than one job.

Parameters:

jobs (list[JobConfig | VirtualJobConfig]) – the jobs to check.

Return type:

None

Returns:

None.

Raises:

ValueError – if a job id is repeated.

_classify_job_annotation(annotation: Any)

Decide whether a type annotation holds a single job or an iterable of jobs.

Looks through Annotated, type aliases (like JobPipeline), and unions.

Parameters:

annotation (Any) – the type annotation of a parameters model’s field.

Returns:

(holds a single job, holds an iterable of jobs)

Return type:

tuple[bool, bool]

Raises:

None. –

Example

from pyantz.infrastructure.config import JobConfig, JobPipeline
from pyantz.infrastructure.virtual.compile import _classify_job_annotation

print(_classify_job_annotation(JobConfig | None))
print(_classify_job_annotation(JobPipeline))
print(_classify_job_annotation(list[str]))

(True, False)
(False, True)
(False, False)

_compile_parameters(jobs: list[JobConfig])

Apply parameter compilation to every job in the list.

Parameters:

jobs (list[JobConfig]) – the concrete jobs.

Returns:

the jobs with the jobs nested in their parameters compiled.

Return type:

list[JobConfig]

Raises:

ValueError – if nested jobs can’t be compiled (see compile_virtual()).

_compile_parameters_singular(job: JobConfig)

Expand job-typed parameters within a single job config.

Some jobs (e.g. if_else, pipeline_expansion) hold other jobs as parameters. This function locates those nested job configs and ensures virtual ones are compiled too.

Parameters:

job (JobConfig) – the concrete job.

Returns:

the job with its nested jobs compiled; unchanged if its function has no registered parameters model.

Return type:

JobConfig

Raises:

ValueError – if nested jobs can’t be compiled (see compile_virtual()).

_compile_singular(job: AnyJobConfig, dep_jobs: list[JobConfig], virtual_ids: Collection[str] = ())

Compile the given job.

Virtual jobs will “consume” this dependencies (they will likely wrap them).

The wrapped jobs no longer see other virtual jobs (those never exist at runtime), so dependencies on other virtual jobs are moved onto the compiled result. The other virtual job then wraps the compiled result when it is compiled, which nests the jobs so they receive the effects of every virtual job they depended on.

Parameters:

• job (JobConfig | VirtualJobConfig) – the job to compile.

• dep_jobs (list[JobConfig]) – the compiled jobs depending on job.

• virtual_ids (Collection[str]) – the ids of every virtual job of the pipeline.

Returns:

for a concrete job, the job followed by dep_jobs; for a virtual job, what it compiles to.

Return type:

list[JobConfig]

Raises:

pydantic.ValidationError – if a virtual job’s parameters are invalid.

_find_joblike_parameters_with_model(job: JobConfig)

Find parameters which appear to be a job.

Returns a tuple of single jobs and iterable job parameters. If not registered, none is returned to signal that this job may have job config parameters, but they aren’t registered so we can’t tell.

Parameters:

job (JobConfig) – the concrete job.

Returns:

the names of the parameters holding a job and of those holding jobs, or None if the job’s function has no registered parameters model.

Return type:

tuple[Iterable[str], Iterable[str]] | None

Raises:

None. –

Example

from pyantz.infrastructure.config import JobConfig
from pyantz.infrastructure.virtual.compile import (
    _find_joblike_parameters_with_model,
)

if_else = "pyantz.jobs.branching.if_else.if_else"
job = JobConfig(function=if_else, parameters={})
single, iterable = _find_joblike_parameters_with_model(job)
print(sorted(single), sorted(iterable))

['if_false', 'if_true'] []

_is_non_empty_iterable(value: object)

Return True for non-empty iterables which aren’t strings or mappings.

Parameters:

value (object) – the value to check.

Returns:

True if value is a non-empty iterable other than str, bytes or a mapping.

Return type:

bool

Raises:

None. –

Example

from pyantz.infrastructure.virtual.compile import _is_non_empty_iterable

print([_is_non_empty_iterable(v) for v in ([1], [], "ab", {"a": 1})])

[True, False, False, False]

_try_to_validate_job(possible_job: Mapping[str, Any] | AnyJobConfig)

Attempt to parse possible_job as a job config; return None on failure.

Used when scanning nested parameters: if validation fails the parameter is not a job dict and should be left alone.

Parameters:

possible_job (Mapping[str, Any] | JobConfig | VirtualJobConfig) – a job config (returned unchanged) or a value which may describe one.

Returns:

the job, or None if the value isn’t one.

Return type:

JobConfig | VirtualJobConfig | None

Raises:

None. –

Example

from pyantz.infrastructure.virtual.compile import _try_to_validate_job

job = {"function": "pyantz.jobs.testing.nop.do_nothing", "parameters": {}}
print(type(_try_to_validate_job(job)).__name__)
print(_try_to_validate_job({"not": "a job"}))

JobConfig
None

compile_virtual(jobs: list[AnyJobConfig])

Compile any virtual jobs to concrete jobs.

The jobs are compiled from the last to the first in dependency order. A virtual job consumes the (compiled) jobs which depend on it, directly or not, and is replaced by what it compiles them into. Jobs nested in the parameters of a job (such as the branches of if_else) are compiled too.

Parameters:

jobs (list[JobConfig | VirtualJobConfig]) – the jobs of a pipeline, virtual or not.

Returns:

the concrete jobs.

Return type:

list[JobConfig]

Raises:

• ValueError – if job ids are repeated, the jobs depend on unknown jobs or have a dependency cycle.

• pydantic.ValidationError – if a virtual job’s parameters are invalid.

Example

from pyantz.infrastructure.config import JobConfig, VirtualJobConfig
from pyantz.infrastructure.virtual import compile_virtual

add_x = VirtualJobConfig(
    job_id="add_x",
    function="pyantz.virtual.add_variables.AddVariables",
    parameters={"variables": {"x": 1}},
)
job = JobConfig(
    depends_on={"add_x"},
    function="pyantz.jobs.testing.nop.do_nothing",
    parameters={},
)
(wrapper,) = compile_virtual([add_x, job])
(inner,) = wrapper.parameters["jobs"]
print(wrapper.function.__name__, inner.depends_on)

run_jobs_in_context set()

Module contents

Virtual jobs edit jobs which depend on them to achieve some desired effect.

For example, rather than a user setting a variable wrapper context, they can use a virtual job “add_variable”. ANy jobs which depend on “add_variable” will, instead of being submitted and actually dependent, be transformed into child jobs of the virtual job, which will submit them with modified context.

In other words,

Add Variable -> JobA

Is actually

SubmitInContext[JobA]