pyantz.infrastructure.config package

pyantz.infrastructure.config package#

Subpackages#

Submodules#

pyantz.infrastructure.config.abstract_job module#

Shared base class for all job configuration types.

AbstractJobConfig is inherited by both JobConfig (concrete jobs) and VirtualJobConfig (compile-time jobs). It provides the common job_id, depends_on, parameters, and strict fields.

class AbstractJobConfig(*, job_id: str = <factory>, depends_on: set[str] | None = None, parameters: ~collections.abc.Mapping[str, ~typing.Any], strict: bool = False)[source]#

Bases: BaseModel

Shared between virtual and real jobs.

Parameters:

job_id (str)
depends_on (Annotated[set[str] | None, WithJsonSchema(json_schema={'type': 'array', 'items': {'type': 'string'}}, mode=None)])
parameters (Mapping[str, Any])
strict (bool)

_abc_impl = <_abc._abc_data object>#

depends_on: set[str] | None#: Ids of the jobs which must finish before this one can run.

job_id: str#: Unique identifier for this job; generated if the user doesn’t give one.

parameters: Mapping[str, Any]#: Parameters to pass to the job while it’s running; strings may hold %{variable} references.

strict: bool#: If true, the parameters are validated when the configuration is loaded, so they can’t depend on variables set while the pipeline runs, and the function must be registered with add_parameters().

str_uuid4()[source]#

Return uuid4 as a string.

Used as the default factory of AbstractJobConfig.job_id.

Parameters:: None.
Returns:: a new random uuid, such as "6f1c...-...".
Return type:: str
Raises:: None. –

Example

import uuid

from pyantz.infrastructure.config.abstract_job import str_uuid4

print(uuid.UUID(str_uuid4()).version)

pyantz.infrastructure.config.abtrast_job module#

Old (misspelled) name of pyantz.infrastructure.config.abstract_job.

Kept so that imports of the old name keep working.

pyantz.infrastructure.config.fn_utils module#

Utilities for (de)serialising Python callables and class references.

Functions are stored in configs as fully-qualified dotted import paths (e.g. "pyantz.jobs.files.simple.copy"). serialize_function() converts a callable back to that string, and import_module_item_by_name() dynamically imports it again at runtime.

ALLOWED_MODULES_VARIABLE = 'PYANTZ_ALLOWED_MODULES'#: Environment variable holding the comma separated module prefixes which configurations may import functions from; see _check_import_allowed().

ImportedCallable#

A field holding a function, given (and serialized) as its full import path.

Example

from collections.abc import Callable

from pydantic import BaseModel

from pyantz.infrastructure.config.fn_utils import ImportedCallable

class Call(BaseModel):
    func: ImportedCallable[Callable[..., object]]

call = Call(func="json.dumps")
print(call.func([1]), call.model_dump())

[1] {'func': 'json.dumps'}

alias of Annotated[_F, BeforeValidator(func=import_function_by_name, json_schema_input_type=PydanticUndefined), PlainSerializer(func=serialize_function, return_type=str, when_used=always), WithJsonSchema(json_schema={‘format’: ‘base64’, ‘type’: ‘string’}, mode=None)]

_check_import_allowed(key: str)[source]#

Refuse names outside of the modules listed in PYANTZ_ALLOWED_MODULES.

The variable holds comma separated module prefixes (pyantz,my_package); a name is allowed if it is one of them or inside one (pyantz.jobs.x for pyantz). Unset or empty, everything is allowed: a configuration is code, and the person who runs it is trusted. Set it wherever configurations come from someone else. The check happens before importing, so modules that are not allowed never run.

Parameters:: key (str) – the full dotted name to be imported.
Return type:: None
Returns:: None.
Raises:: ValueError – if the name is not allowed

Example

import os

from pyantz.infrastructure.config.fn_utils import _check_import_allowed

os.environ["PYANTZ_ALLOWED_MODULES"] = "pyantz, my_package"
try:
    _check_import_allowed("pyantz.jobs.files.simple.copy")  # allowed
    _check_import_allowed("os.system")
except ValueError:
    print("os.system is not allowed")
finally:
    del os.environ["PYANTZ_ALLOWED_MODULES"]

os.system is not allowed

import_function_by_name(fn_path: Any)[source]#

Import a function by its name.

Used to validate the function of a job; a value which isn’t a string (such as the function itself) is returned unchanged.

Parameters:: fn_path (Any) – the function’s full dotted import path, or the function.
Returns:: the imported function.
Return type:: JobFunctionType
Raises:: ValueError – if the name can’t be imported or isn’t allowed by ALLOWED_MODULES_VARIABLE.

Example

from pyantz.infrastructure.config.fn_utils import import_function_by_name

fn = import_function_by_name("pyantz.jobs.testing.nop.do_nothing")
print(fn.__name__)

do_nothing

import_module_item_by_name(key: Any)[source]#

Import a parameters model by its name.

Any item of a module can be imported, including attributes of classes such as package.module.Class.method: the longest prefix of the name which is a module is imported, and the rest is looked up on it.

Parameters:: key (Any) – the item’s full dotted import path; anything which isn’t a string is returned unchanged.
Returns:: the imported item.
Return type:: Any
Raises:: ValueError – if the name has fewer than two parts, its module can’t be imported, the module lacks the item, or the name isn’t allowed by ALLOWED_MODULES_VARIABLE.

Example

from pyantz.infrastructure.config.fn_utils import (
    import_module_item_by_name,
)

print(import_module_item_by_name("pathlib.Path.home").__qualname__)
try:
    import_module_item_by_name("pathlib.Missing")
except ValueError as exc:
    print(exc)

Path.home
Cannot import 'pathlib.Missing': Module 'pathlib' has no attribute 'Missing'

serialize_function(fn: Callable[..., Any])[source]#

Serialize the function so that others can import it dynamically.

Parameters:: fn (Callable[..., Any]) – the function (or class) to serialize.
Returns:: the full dotted import path of fn.
Return type:: str
Raises:: ValueError – if the function can’t be imported by name (lambdas, functions defined inside other functions, and functions without a module, such as those made by code run with exec)

Example

from pyantz.infrastructure.config.fn_utils import serialize_function
from pyantz.jobs.files.simple import copy

print(serialize_function(copy))
try:
    serialize_function(lambda: None)
except ValueError:
    print("lambdas can't be imported by name")

pyantz.jobs.files.simple.copy
lambdas can't be imported by name

pyantz.infrastructure.config.graph_validation module#

Check the structure of a set of jobs: ids, dependencies and cycles.

find_graph_problems() reports every problem at once, so that a user can fix a configuration in one go instead of one error at a time.

find_graph_problems(jobs: Iterable[AbstractJobConfig])[source]#

List everything wrong with the ids and dependencies of the jobs.

Parameters:: jobs (Iterable[AbstractJobConfig]) – the jobs of one pipeline (virtual jobs included)
Returns:: a message for each problem; empty if the jobs are fine
Return type:: list[str]
Raises:: None – problems are reported in the returned list.

Example

from pyantz.infrastructure.config import JobConfig
from pyantz.infrastructure.config.graph_validation import (
    find_graph_problems,
)

nop = "pyantz.jobs.testing.nop.do_nothing"
jobs = [
    JobConfig(job_id="a", depends_on={"b"}, function=nop, parameters={}),
    JobConfig(job_id="b", depends_on={"a"}, function=nop, parameters={}),
    JobConfig(job_id="c", depends_on={"x"}, function=nop, parameters={}),
]
for problem in find_graph_problems(jobs):
    print(problem)

Job 'c' depends on unknown job 'x'
Dependency cycle: a -> b -> a

pyantz.infrastructure.config.host_config module#

Configuration for where the pipeline controller runs.

The host config controls where pyantz starts its submitter process, which is distinct from how individual jobs are submitted (the runner config).

LocalConfig — run the controller on this machine (default).
SshConfig — SSH into a remote host, copy the config, and run pyantz there.
ContainerConfig — exec into a running Docker/Podman container.

After reaching the remote host, the subsequent_host field lets you chain hops (e.g. SSH into a bastion, then exec into a container).

class ContainerConfig(*, environment_variables_file: str | None = None, output_dir: str | None = None, working_dir: str | None = None, requirements: list[str] | None = None, copy_project_dir: bool = False, subsequent_host: HostConfig = LocalConfig(type_='local'), environment_variables: Mapping[str, str]=<factory>, type_: Literal['container'] = 'container', container_service: Literal['podman', 'docker']='docker', image: str = 'ghcr.io/astral-sh/uv:alpine', name: str | None = None)[source]#

Bases: RemoteConfig

Execute in a docker container that is running on the machine.

Parameters:

environment_variables_file (str | None)
output_dir (str | None)
working_dir (str | None)
requirements (list[str] | None)
copy_project_dir (bool)
subsequent_host (HostConfig)
environment_variables (Mapping[str, str])
type_ (Literal['container'])
container_service (Literal['podman', 'docker'])
image (str)
name (str | None)

_abc_impl = <_abc._abc_data object>#

container_service: Literal['podman', 'docker']#: Which container software to use.

image: str#: Image to run if no container is named; it must provide uv.

name: str | None#: Name of the running container to exec into; if not set, a new container is started from image.

type_: Literal['container']#: Identifies this host configuration.

type HostConfig = Annotated[SshConfig | ContainerConfig | LocalConfig, FieldInfo(annotation=NoneType, required=True, discriminator='type_')]#: Any host configuration, told apart by its type_.

class LocalConfig(*, type_: Literal['local'] = 'local')[source]#

Bases: BaseModel

Execute the submitter locally.

Parameters:: type_ (Literal['local'])

_abc_impl = <_abc._abc_data object>#

type_: Literal['local']#: Identifies this host configuration.

class RemoteConfig(*, environment_variables_file: str | None = None, output_dir: str | None = None, working_dir: str | None = None, requirements: list[str] | None = None, copy_project_dir: bool = False, subsequent_host: HostConfig = LocalConfig(type_='local'), environment_variables: Mapping[str, str]=<factory>)[source]#

Bases: BaseModel

Execute on a remote machine.

Just a base class for other configurations.

Parameters:

environment_variables_file (str | None)
output_dir (str | None)
working_dir (str | None)
requirements (list[str] | None)
copy_project_dir (bool)
subsequent_host (HostConfig)
environment_variables (Mapping[str, str])

_abc_impl = <_abc._abc_data object>#

copy_project_dir: bool#: If true, copy this project directory to the remote; primarily useful for testing.

environment_variables: Mapping[str, str]#: Environment variables to set for the command run on the remote host (for containers, in the container).

environment_variables_file: str | None#: If set, a file on this machine with KEY=VALUE lines (blank lines and # comments are ignored) which is read and sent as environment variables of the remote command. Variables in environment_variables take precedence.

output_dir: str | None#: If set, the working directory from the remote will be copied to this directory.

requirements: list[str] | None#: Python requirements to install on the remote (as given to pip install).

subsequent_host: HostConfig#: After connecting to the remote, what host to use there.

working_dir: str | None#: If set, the directory to run in on the remote; if not set, ssh hosts use a new temporary directory created on the remote and containers use /workspace.

class SshConfig(*, environment_variables_file: str | None = None, output_dir: str | None = None, working_dir: str | None = None, requirements: list[str] | None = None, copy_project_dir: bool = False, subsequent_host: HostConfig = LocalConfig(type_='local'), environment_variables: Mapping[str, str]=<factory>, type_: Literal['ssh'] = 'ssh', remote_host: str, remote_port: int = 22, remote_user: str = <factory>, connect_timeout: float = 30, key_filename: str | None = None, known_hosts_policy: Literal['reject', 'warn', 'auto_add']='reject', run_async: bool = True, sync_data_at_end: bool = False)[source]#

Bases: RemoteConfig

Execute using ssh to access a remote machine.

Parameters:

environment_variables_file (str | None)
output_dir (str | None)
working_dir (str | None)
requirements (list[str] | None)
copy_project_dir (bool)
subsequent_host (HostConfig)
environment_variables (Mapping[str, str])
type_ (Literal['ssh'])
remote_host (str)
remote_port (int)
remote_user (str)
connect_timeout (float)
key_filename (str | None)
known_hosts_policy (Literal['reject', 'warn', 'auto_add'])
run_async (bool)
sync_data_at_end (bool)

_abc_impl = <_abc._abc_data object>#

check_syncs()[source]#

Check that sync settings make sense.

Parameters:: None.
Returns:: the validated configuration.
Return type:: Self
Raises:: ValueError – if sync_data_at_end is set with run_async or without output_dir.

Example

from pydantic import ValidationError

from pyantz.infrastructure.config import SshConfig

try:
    SshConfig(remote_host="hpc", sync_data_at_end=True)
except ValidationError:
    print("can't sync data at the end of an async run")

can't sync data at the end of an async run

connect_timeout: float#: Seconds to wait when connecting before giving up.

key_filename: str | None#: Private key file to authenticate with; if not set, paramiko tries the ssh agent and the usual keys in ~/.ssh.

known_hosts_policy: Literal['reject', 'warn', 'auto_add']#

reject refuses to connect, warn connects and logs a warning, auto_add trusts and remembers it (unsafe on untrusted networks).

Type:: What to do with a host that is not in the known hosts

remote_host: str#: Host to ssh into.

remote_port: int#: Port to use for ssh.

remote_user: str#: User to log in as; defaults to the current user, looked up when the config is created.

run_async: bool#: If true, don’t wait for the command to finish before closing the ssh connection.

sync_data_at_end: bool#: If true, copy the remote working directory back to output_dir once the command finishes; needs run_async to be false.

type_: Literal['ssh']#: Identifies this host configuration.

_current_user()[source]#

Get the name of the user running pyantz.

Used as the default of SshConfig.remote_user.

Parameters:: None.
Returns:: the login name of the current user.
Return type:: str
Raises:: OSError – if the user’s name can’t be found (see getpass.getuser()).

pyantz.infrastructure.config.initial module#

Top-level pipeline configuration.

InitialConfig is the root object that a user creates (in JSON or Python) to describe a complete pipeline run. It ties together the job list, the submitter (local or SLURM), optional global variables, and the host on which the submitter should be started.

Example of InitialConfig:

from pyantz.infrastructure.config import InitialConfig

config = InitialConfig.model_validate(
    {
        "jobs": [
            {
                "function": "pyantz.jobs.testing.nop.do_nothing",
                "parameters": {},
            },
        ],
        "submitter": {"type_": "local_proc", "working_directory": "."},
        "variables": {"data_dir": "/data"},
    }
)
print(config.host.type_, config.submitter.type_, len(config.jobs))

local local_proc 1

type AnyRunner = Annotated[LocalRunnerConfig | SlurmRunnerConfig, FieldInfo(annotation=NoneType, required=True, discriminator='type_')]#: Any submitter configuration, told apart by its type_.

class InitialConfig(*, jobs: tuple[~pyantz.infrastructure.config.job.JobConfig, ...], submitter: ~pyantz.infrastructure.config.initial.AnyRunner, variables: ~collections.abc.Mapping[str, ~typing.Any] = <factory>, host: ~pyantz.infrastructure.config.host_config.HostConfig = LocalConfig(type_='local'))[source]#

Bases: BaseModel

Configuration of the overall system.

Passed by the user and used to setup the system. This is the primary user configuration.

Parameters:

jobs (Annotated[tuple[JobConfig, ...], BeforeValidator(func=~pyantz.infrastructure.config.initial.make_and_compile, json_schema_input_type=PydanticUndefined)])
submitter (AnyRunner)
variables (Mapping[str, Any])
host (HostConfig)

_abc_impl = <_abc._abc_data object>#

_check_dependencies_exist()[source]#

Ensure every job only depends on jobs in this configuration.

Parameters:: None.
Returns:: the validated configuration.
Return type:: Self
Raises:: ValueError – if a job depends on a job id which isn’t in the configuration.

host: HostConfig#

this machine, over ssh or in a container.

Type:: Where the submitter runs

jobs: tuple[JobConfig, ...]#: The jobs to run, with virtual jobs compiled away.

submitter: AnyRunner#

local_proc or slurm_sbatch.

Type:: How the jobs are submitted

variables: Mapping[str, Any]#: Variables available to every job.

with_new_host(new_host: HostConfig)[source]#

Return this configuration with a new host setup.

Parameters:: new_host (HostConfig) – the host configuration to use.
Returns:: a copy of this configuration with host replaced.
Return type:: Self
Raises:: None. –

Example

from pyantz.infrastructure.config import InitialConfig, LocalConfig

config = InitialConfig.model_validate(
    {
        "jobs": [],
        "submitter": {"type_": "local_proc", "working_directory": "."},
        "host": {"type_": "container", "name": "builder"},
    }
)
print(config.with_new_host(LocalConfig()).host.type_)

local

make_and_compile(jobs: Any)[source]#

Validate raw job dicts and compile virtual jobs into concrete ones.

Used as the BeforeValidator for InitialConfig.jobs. First calls make_job() on each element to produce typed job configs, then passes the result through compile_virtual() to expand any virtual jobs.

Parameters:

jobs (Any) – An iterable of raw job-config dicts or pre-validated job config objects.

Return type:

list[JobConfig]

Returns:

Flat list of compiled JobConfig instances with all virtual jobs expanded.

Raises:

ValueError – if the jobs have duplicate ids, unknown dependencies or a dependency cycle, or a job is invalid.
TypeError – if a job isn’t a mapping or job config.

Example

from pyantz.infrastructure.config.initial import make_and_compile

jobs = make_and_compile(
    [
        {
            "function": "pyantz.virtual.add_variables.AddVariables",
            "parameters": {"variables": {"x": 1}},
            "job_id": "vars",
        },
        {
            "function": "pyantz.jobs.testing.nop.do_nothing",
            "parameters": {},
            "depends_on": ["vars"],
        },
    ]
)
print([job.function.__name__ for job in jobs])

['run_jobs_in_context']

pyantz.infrastructure.config.job module#

Configuration of a single job to be run.

Example of JobConfig:

from pyantz.infrastructure.config import JobConfig

job = JobConfig.model_validate(
    {
        "name": "say hi",
        "function": "pyantz.jobs.testing.nop.do_nothing",
        "parameters": {},
        "max_retries": 2,
    }
)
print(job.function.__name__, job.max_retries, job.dependency_type)

do_nothing 2 afterok

class JobConfig(*, job_id: str = <factory>, depends_on: set[str] | None = None, parameters: ~collections.abc.Mapping[str, ~typing.Any], strict: bool = False, name: str | None = None, function: ~pyantz.infrastructure.config.job.JobFunctionType, num_attempted_runs: int = 0, timeout_seconds: float | None = None, max_retries: int = 0, dependency_type: ~typing.Literal['afterok', 'afterany', 'afternotok'] = 'afterok', virtual: ~typing.Literal[False] = False)[source]#

Bases: AbstractJobConfig

A job to be run.

Parameters:

job_id (str)
depends_on (Annotated[set[str] | None, WithJsonSchema(json_schema={'type': 'array', 'items': {'type': 'string'}}, mode=None)])
parameters (Mapping[str, Any])
strict (bool)
name (str | None)
function (Annotated[JobFunctionType, BeforeValidator(func=~pyantz.infrastructure.config.fn_utils.import_function_by_name, json_schema_input_type=PydanticUndefined), WithJsonSchema(json_schema={'type': 'string', 'format': 'base64'}, mode=None)])
num_attempted_runs (int)
timeout_seconds (Annotated[float | None, Gt(gt=0)])
max_retries (Annotated[int, Ge(ge=0)])
dependency_type (Literal['afterok', 'afterany', 'afternotok'])
virtual (Literal[False])

_abc_impl = <_abc._abc_data object>#

dependency_type: Literal['afterok', 'afterany', 'afternotok']#: How this job waits for the jobs it depends on (SLURM runner only; the local runner always waits for success): afterok runs it if they succeeded, afterany after they finished either way (e.g., cleanup jobs), afternotok only if one failed.

function: JobFunctionType#: Function to actually run for this job, given by its full import path.

max_retries: int#: How many times a failed job is run again before it is given up on. Retried jobs run again from the start, so jobs which submit other jobs should not be retried (the children would be submitted twice).

name: str | None#: Human readable name for this job.

num_attempted_runs: int#: How many times the job has been run, so the runner can decide whether to try it again.

serialize_job_function(fn: JobFunctionType)[source]#

Serialize the function.

Parameters:: fn (JobFunctionType) – the job’s function.
Returns:: the function’s full import path.
Return type:: str
Raises:: ValueError – if the function can’t be imported by name, e.g., a lambda.

Example

from pyantz.infrastructure.config import JobConfig

job = JobConfig(
    function="pyantz.jobs.testing.nop.do_nothing", parameters={}
)
print(job.model_dump()["function"])

pyantz.jobs.testing.nop.do_nothing

timeout_seconds: float | None#: If set, the local runner stops the job (and marks it as failed) after it has been running this many seconds. The job then runs in its own process.

validate_fn()[source]#

Check the parameters for the function.

See check_config() for when the parameters are checked.

Parameters:: None.
Returns:: the validated job.
Return type:: Self
Raises:: ValueError – if the parameters are checked and don’t fit the function’s parameters model, or the job is strict and its function isn’t registered.

Example

from pydantic import ValidationError

from pyantz.infrastructure.config import JobConfig

try:
    JobConfig(
        function="pyantz.jobs.files.simple.copy",
        parameters={"src": "a.txt"},  # dst is missing
        strict=True,
    )
except ValidationError:
    print("bad parameters")

bad parameters

virtual: Literal[False]#: Denotes that this a concrete job; makes introspection easier/faster at runtime.

type JobFunctionType = Callable[[ParametersType, SubmissionFnType], bool]#: Function that performs the job expected.

class JobWithContext(*, job_id: str = <factory>, depends_on: set[str] | None = None, parameters: ~collections.abc.Mapping[str, ~typing.Any], strict: bool = False, name: str | None = None, function: ~pyantz.infrastructure.config.job.JobFunctionType, num_attempted_runs: int = 0, timeout_seconds: float | None = None, max_retries: int = 0, dependency_type: ~typing.Literal['afterok', 'afterany', 'afternotok'] = 'afterok', virtual: ~typing.Literal[False] = False, variables: ~collections.abc.Mapping[str, ~typing.Any] | None = None)[source]#

Bases: JobConfig

Job to be run along with its variables in scope.

Parameters:

job_id (str)
depends_on (Annotated[set[str] | None, WithJsonSchema(json_schema={'type': 'array', 'items': {'type': 'string'}}, mode=None)])
parameters (Mapping[str, Any])
strict (bool)
name (str | None)
function (Annotated[JobFunctionType, BeforeValidator(func=~pyantz.infrastructure.config.fn_utils.import_function_by_name, json_schema_input_type=PydanticUndefined), WithJsonSchema(json_schema={'type': 'string', 'format': 'base64'}, mode=None)])
num_attempted_runs (int)
timeout_seconds (Annotated[float | None, Gt(gt=0)])
max_retries (Annotated[int, Ge(ge=0)])
dependency_type (Literal['afterok', 'afterany', 'afternotok'])
virtual (Literal[False])
variables (Mapping[str, Any] | None)

_abc_impl = <_abc._abc_data object>#

classmethod _update_ids_in_hyper_parameters(data: Any)[source]#

Update ids in the “hyper params” (like id, name, depends_on).

This happens before validation, so the resolved values are validated like any value given directly.

Parameters:: data (Any) – the raw data being validated.
Returns:: the data with variables resolved in job_id, name and depends_on; anything which isn’t a mapping is returned unchanged.
Return type:: Any
Raises:: ValueError – if a variable refers to itself (directly or not).

Example

from pyantz.infrastructure.config import JobWithContext

job = JobWithContext.model_validate(
    {
        "name": "fit_%{model}",
        "function": "pyantz.jobs.testing.nop.do_nothing",
        "parameters": {},
        "variables": {"model": "linear"},
    }
)
print(job.name)

fit_linear

classmethod from_config(config_possibly_without_context: JobConfig | VirtualJobConfig)[source]#

Create a JobWithContext from another job config.

The other job config may be the parent class or could be an instance of this class.

Parameters:: config_possibly_without_context (JobConfig | VirtualJobConfig) – some configuration
Returns:: job as defined, with default variables if not present
Return type:: JobWithContext
Raises:: pydantic.ValidationError – if the configuration isn’t a valid job.

Example

from pyantz.infrastructure.config import JobConfig, JobWithContext

job = JobConfig(
    function="pyantz.jobs.testing.nop.do_nothing", parameters={}
)
with_context = JobWithContext.from_config(job)
print(with_context.job_id == job.job_id, with_context.variables)

True None

inherit_context(parent_context: Mapping[str, Any])[source]#

Create a JobWithContext inheriting parent variables.

Parameters:: parent_context (Mapping[str, Any]) – the variables of the parent; this job’s own variables take precedence over them.
Returns:: a copy of this job with the combined variables.
Return type:: JobWithContext
Raises:: None. –

Example

from pyantz.infrastructure.config import JobWithContext

job = JobWithContext(
    function="pyantz.jobs.testing.nop.do_nothing",
    parameters={},
    variables={"a": "mine"},
)
print(job.inherit_context({"a": "parent", "b": "parent"}).variables)

{'a': 'mine', 'b': 'parent'}

variables: Mapping[str, Any] | None#: The variables in scope of the job, used to fill in %{name} references in its parameters (and in its job_id, name and depends_on).

type ParametersType = Mapping[str, Any]#: Jobs are given parameters by the user.

type SubmissionFnType = Callable[[JobConfig], None]#: Function a “user” function receives to submit jobs.

make_job(config: Any)[source]#

Make a job based on a user provided configuration.

The job is virtual if its virtual key says so or, without one, if its function is marked with mark_virtual().

Parameters:

config (Any) – a job configuration (returned unchanged) or a mapping describing one.

Returns:

the validated job.

Return type:

JobConfig | VirtualJobConfig

Raises:

TypeError – if config isn’t a mapping or job configuration.
ValueError – if config has no function, or it can’t be imported.
pydantic.ValidationError – if config isn’t a valid job.

Example

from pyantz.infrastructure.config.job import make_job

job = make_job(
    {
        "function": "pyantz.virtual.add_variables.AddVariables",
        "parameters": {"variables": {}},
    }
)
print(type(job).__name__)

VirtualJobConfig

pyantz.infrastructure.config.pipeline module#

Configuration for pipelines — ordered sequences of jobs.

A JobPipeline is the type alias used wherever a list of jobs can appear in parameters. At parse time Pydantic calls _make_pipeline() to convert a raw list of dicts into typed job config objects.

type JobPipeline = Annotated[list[JobConfig | VirtualJobConfig], BeforeValidator(func=_make_pipeline, json_schema_input_type=PydanticUndefined)]#: A list of jobs, given as job configs or the mappings describing them.

_make_pipeline(pipeline: Any)[source]#

Convert a raw iterable into a validated list of job configs.

Called automatically by Pydantic’s BeforeValidator when a JobPipeline field is populated. Each element is passed to make_job(), which selects JobConfig vs VirtualJobConfig based on the virtual flag or the type of the function value.

Parameters:: pipeline (Any) – Any iterable of raw job-config dicts or pre-validated job config objects.
Return type:: list[JobConfig | VirtualJobConfig]
Returns:: A list of validated job config objects ready for compilation or execution.
Raises:: TypeError – If pipeline is not iterable.

Example

from pyantz.infrastructure.config.pipeline import _make_pipeline

jobs = _make_pipeline(
    [{"function": "pyantz.jobs.testing.nop.do_nothing", "parameters": {}}]
)
print(type(jobs[0]).__name__)

JobConfig

pyantz.infrastructure.config.virtual module#

Configuration of a virtual job.

A virtual job only exists while a configuration is compiled: it is replaced by the concrete jobs its parameters model compiles to, usually wrapping the jobs which depend on it. See pyantz.infrastructure.virtual.

Example of VirtualJobConfig:

from pyantz.infrastructure.config import JobConfig, VirtualJobConfig

virtual_job = VirtualJobConfig(
    function="pyantz.virtual.add_variables.AddVariables",
    parameters={"variables": {"x": 1}},
)
nop = "pyantz.jobs.testing.nop.do_nothing"
job = JobConfig(function=nop, parameters={})
(compiled,) = virtual_job.compile_virtual([job])
print(compiled.function.__name__, compiled.parameters["shared_variables"])

run_jobs_in_context {'x': 1}

class VirtualJobConfig(*, job_id: str = <factory>, depends_on: set[str] | None = None, parameters: ~collections.abc.Mapping[str, ~typing.Any], strict: bool = False, name: str | None = None, function: type[~pyantz.infrastructure.config.virtual.VirtualParamModel], virtual: ~typing.Literal[True] = True)[source]#

Bases: AbstractJobConfig

A job which is only used to create an ordered set of ‘real’ jobs.

Parameters:

job_id (str)
depends_on (Annotated[set[str] | None, WithJsonSchema(json_schema={'type': 'array', 'items': {'type': 'string'}}, mode=None)])
parameters (Mapping[str, Any])
strict (bool)
name (str | None)
function (Annotated[type[VirtualParamModel], AfterValidator(func=~pyantz.infrastructure.config.virtual._is_virtual), BeforeValidator(func=~pyantz.infrastructure.config.fn_utils.import_module_item_by_name, json_schema_input_type=PydanticUndefined)])
virtual (Literal[True])

_abc_impl = <_abc._abc_data object>#

compile_virtual(dependent_jobs: list[JobConfig])[source]#

Compile from our parameters to real jobs.

Parameters:: dependent_jobs (list[JobConfig]) – the (compiled) jobs which depend on this job.
Returns:: the concrete jobs replacing this job and dependent_jobs.
Return type:: list[JobConfig]
Raises:: pydantic.ValidationError – if the parameters are invalid.

property compiled_params: VirtualParamModel#

Build our parameters.

Parameters:: None.
Returns:: the parameters, validated by function.
Return type:: VirtualParamModel
Raises:: pydantic.ValidationError – if the parameters are invalid.

function: type[VirtualParamModel]#: Import path of the virtual job’s parameters model, which must be marked with mark_virtual().

name: str | None#: Human readable name.

virtual: Literal[True]#: Denotes that this is a virtual job.

class VirtualParamModel(*args, **kwargs)[source]#

Bases: Protocol

Model to be inherited by parameters of virtual jobs.

Virtual jobs have no functions, so they are purely defined by these parameter models, which will compile into a real set of job configs (which will primarily focus on the parameters).

_abc_impl = <_abc._abc_data object>#

_is_protocol = True#

_is_runtime_protocol = True#

abstractmethod compile_virtual(dependent_jobs: list[JobConfig])[source]#

Compile our job into a list of real jobs.

Parameters:: dependent_jobs (list[JobConfig]) – the (compiled) jobs which depend on the virtual job.
Returns:: the concrete jobs replacing the virtual job and dependent_jobs.
Return type:: list[JobConfig]
Raises:: Exception – implementations may raise if they can’t compile.

abstractmethod classmethod model_validate(params: Mapping[str, Any])[source]#

Pydantic model validation.

Parameters:: params (Mapping[str, Any]) – the virtual job’s parameters.
Returns:: the validated parameters.
Return type:: Self
Raises:: pydantic.ValidationError – if the parameters are invalid.

_is_virtual(params: VirtualParamModel)[source]#

Check that a virtual job’s function is marked as virtual.

Parameters:: params (VirtualParamModel) – the parameters model given as the job’s function.
Returns:: params unchanged.
Return type:: VirtualParamModel
Raises:: ValueError – if params isn’t marked with mark_virtual().

Module contents#

Configuration passed by the user to run a pipeline.

pyantz.infrastructure.config package

Contents

pyantz.infrastructure.config package#

Subpackages#

Submodules#

pyantz.infrastructure.config.abstract_job module#

pyantz.infrastructure.config.abtrast_job module#

pyantz.infrastructure.config.fn_utils module#

pyantz.infrastructure.config.graph_validation module#

pyantz.infrastructure.config.host_config module#

pyantz.infrastructure.config.initial module#

pyantz.infrastructure.config.job module#

pyantz.infrastructure.config.pipeline module#

pyantz.infrastructure.config.virtual module#

Module contents#