pyantz.virtual package

Submodules

pyantz.virtual.add_variables module

Virtual job that injects variables into its dependent jobs.

AddVariables is a virtual job — it compiles away at startup and leaves no footprint at runtime. Use it in a pipeline to declare shared variables that should be visible to all subsequent jobs.

Example JSON usage:

{
  "function": "pyantz.virtual.add_variables.AddVariables",
  "virtual": true,
  "parameters": {
    "variables": {"data_dir": "/mnt/data", "batch_size": 64}
  }
}

Example of AddVariables:

import os

from pyantz import start

start(
    {
        "submitter": {
            "type_": "local_proc",
            "working_directory": ".",
            "use_same_proc": True,
        },
        "jobs": [
            {
                "job_id": "set_name",
                "function": "pyantz.virtual.add_variables.AddVariables",
                "parameters": {"variables": {"name": "results"}},
            },
            {
                "function": "pyantz.jobs.files.simple.mkdir",
                "parameters": {"dir_path": "%{name}"},
                "depends_on": ["set_name"],
            },
        ],
    }
)
print(os.path.isdir("results"))

True

class AddVariables(*, variables: Mapping[str, Any])

Bases: BaseModel

Parameters to add a variable.

Parameters:

variables (Mapping[str, Any])

_abc_impl = <_abc._abc_data object>

compile_virtual(deps: list[JobConfig])

Wrap deps in a run_jobs_in_context job.

Produces a single concrete job that, when executed, injects variables into every job in deps before submitting them.

Parameters:

deps (list[JobConfig]) – The list of downstream jobs that should receive the new variables as inherited context.

Return type:

list[JobConfig]

Returns:

A single-element list containing the compiled JobConfig.

Raises:

pydantic.ValidationError – if the compiled job is invalid.

Example

from pyantz.infrastructure.config import JobConfig
from pyantz.virtual.add_variables import AddVariables

job = JobConfig(
    function="pyantz.jobs.testing.nop.do_nothing", parameters={}
)
(wrapper,) = AddVariables(variables={"x": 1}).compile_virtual([job])
print(wrapper.parameters["jobs"][0].job_id == job.job_id)

True

variables: Mapping[str, Any]

Variables to add to the jobs which depend on this one.

Module contents

Holds code to define a virtual job.