pyantz.infrastructure.config.parameters package

Submodules

pyantz.infrastructure.config.parameters.compile_check module

Parameter validation performed when a config is first loaded.

check_config() decides whether to validate a job’s parameters against its registered Pydantic model at startup time (before any job runs). This is distinct from runtime validation, which happens inside add_parameters().

Startup checking is disabled by default to allow template variables to be resolved at runtime; set strict: true on a job or check_at_startup: true on its decorator to force early validation.

_check_params(job_config: JobConfig, *, strict: bool = False)

Check the parameters of the job configuration.

Parameters:

• job_config (JobConfig) – the job to check; its variables, if any, are resolved first.

• strict (bool) – if True, parameters with unresolved variables must be valid too.

Returns:

True if the parameters are valid, or (when not strict) invalid but holding variables which may be set later; False otherwise, including when the function has no parameters model.

Return type:

bool

Raises:

ValueError – if resolving the variables finds a recursive reference.

check_config(job_config: JobConfig, *, strict: bool = False, check_at_startup: bool = False)

Enforce type checked parameters on the config.

If variables in the job config (JobWithContext), it will resolve those first. The parameters are only checked if check_at_startup is set or the function was registered with check_at_startup=True; otherwise they are checked when the job runs.

Parameters:

• job_config (JobConfig) – the job to check.

• strict (bool) – if True, the function must be registered with add_parameters(), and parameters with unresolved variables must already be valid.

• check_at_startup (bool) – if True, check the parameters now even if the function doesn’t ask for it.

Returns:

True if the job is acceptable, False if it isn’t.

Return type:

bool

Raises:

ValueError – if resolving the variables finds a recursive reference.

Example

from pyantz.infrastructure.config import JobConfig
from pyantz.infrastructure.config.parameters.compile_check import (
    check_config,
)

job = JobConfig(function="pyantz.jobs.files.simple.copy", parameters={})
print(check_config(job))  # checked when the job runs instead
print(check_config(job, check_at_startup=True))  # src and dst missing

True
False

get_params(job: JobConfig)

Retrieve parameters attached to the job function, if any registered.

Parameters:

job (JobConfig) – the job whose function to look at.

Returns:

the parameters model the function was registered with by add_parameters(), or None.

Return type:

type[BaseModel] | None

Raises:

None. –

Example

from pyantz.infrastructure.config import JobConfig
from pyantz.infrastructure.config.parameters import get_params

job = JobConfig(
    function="pyantz.jobs.files.simple.copy",
    parameters={"src": "a", "dst": "b"},
)
print(get_params(job).__name__)

CopyParams

pyantz.infrastructure.config.parameters.decorators module

Decorate functions and strongly enforce parameter types.

PYANTZ_REGISTERED_FUNCTION: Final[dict[str, JobFunctionType]] = {'pyantz.jobs.analysis.external.external_analysis': <function external_analysis>, 'pyantz.jobs.analysis.external.external_simple': <function external_simple>, 'pyantz.jobs.analysis.external.save_from_external_extraction': <function save_from_external_extraction>, 'pyantz.jobs.analysis.simple.columnar_operation': <function columnar_operation>, 'pyantz.jobs.analysis.simple.join_parquets': <function join_parquets>, 'pyantz.jobs.branching.case_matrix.create_case_matrix': <function create_case_matrix>, 'pyantz.jobs.branching.case_matrix.pipeline_expansion': <function pipeline_expansion>, 'pyantz.jobs.branching.case_matrix.pipeline_expansion_with_output_dir': <function pipeline_expansion_with_output_dir>, 'pyantz.jobs.branching.if_else.conditionally_execute': <function conditionally_execute>, 'pyantz.jobs.branching.if_else.if_else': <function if_else>, 'pyantz.jobs.files.check.file_contains': <function file_contains>, 'pyantz.jobs.files.check.file_exists': <function file_exists>, 'pyantz.jobs.files.edit.edit_external_method': <function edit_external_method>, 'pyantz.jobs.files.edit.edit_json': <function edit_json>, 'pyantz.jobs.files.edit.edit_yaml': <function edit_yaml>, 'pyantz.jobs.files.moving.copy': <function copy>, 'pyantz.jobs.files.moving.move': <function move>, 'pyantz.jobs.files.moving.rsync': <function rsync>, 'pyantz.jobs.files.reformat.csv_to_parquet': <function csv_to_parquet>, 'pyantz.jobs.files.reformat.excel_to_csv': <function excel_to_csv>, 'pyantz.jobs.files.simple.copy': <function copy>, 'pyantz.jobs.files.simple.create': <function create>, 'pyantz.jobs.files.simple.delete': <function delete>, 'pyantz.jobs.files.simple.mkdir': <function mkdir>, 'pyantz.jobs.files.simple.move': <function move>, 'pyantz.jobs.subproc.dispatch.dispatch': <function dispatch>, 'pyantz.jobs.testing.touch_file.touch_file': <function touch_file>, 'pyantz.jobs.wrappers.local.run_pipeline': <function run_pipeline>, 'pyantz.jobs.wrappers.loops.local_for_loop': <function local_for_loop>, 'pyantz.jobs.wrappers.submit.new_host': <function new_host>, 'pyantz.jobs.wrappers.submit.submit': <function submit>, 'pyantz.jobs.wrappers.variables.run_jobs_in_context': <function run_jobs_in_context>, 'pyantz.jobs.wrappers.variables.run_jobs_with_temp_dir': <function run_jobs_with_temp_dir>, 'pyantz.jobs.wrappers.variables.set_variables': <function set_variables>}

Every function registered by add_parameters(), keyed by its full name, so registering a function again (a module reload) replaces it.

VIRTUAL_MARKER: Final[str] = '__PYANTZ_VIRTUAL__'

Attribute set by mark_virtual() on virtual job parameter models.

_name_matches(registered_name: str, query: str)

Check if the query is the whole name or a trailing part split on dots.

Parameters:

• registered_name (str) – the full dotted name of a registered function.

• query (str) – the name searched for.

Returns:

True if query is registered_name or its trailing parts.

Return type:

bool

Raises:

None. –

Example

from pyantz.infrastructure.config.parameters.decorators import (
    _name_matches,
)

print(_name_matches("a.simple.copy", "simple.copy"))
print(_name_matches("a.simple.copy", "py"))

True
False

add_parameters[T: BaseModel](param_cls: type[T], *, check_at_startup: bool = False)

Use the provided pydantic model to type check the parameters.

Also registers the functions in a global static function list for the API

At runtime, it will cast the parameters to the provided parameter class and pass that pydantic model to the wrapped function. Useful to enforce strict typing in a function.

Parameters:

• param_cls (type[T]) – basemodel to use to validate

• check_at_startup (bool) – if True, will check the parameters class at startup. this disallows variables as those won’t be set yet. Instead, prefer setting this value in the configuration as part of the initial params.

Returns:

the decorator, which returns the registered job function.

Return type:

Callable[[Callable[[T, SubmissionFnType], bool]], JobFunctionType]

Raises:

None. –

Example

from pydantic import BaseModel

from pyantz.infrastructure.config import add_parameters

class GreetParams(BaseModel):
    name: str
    times: int = 1

@add_parameters(GreetParams)
def greet(params, submit_fn):
    print(" ".join([f"hello {params.name}"] * params.times))
    return True

greet({"name": "ant", "times": 2}, lambda job: None)
print(greet.PYANTZ_VALIDATION_MODEL.__name__)

hello ant hello ant
GreetParams

get_parameters(fn: Any)

Get the add parameters if PYANTZ_VALIDATION_MODEL set.

Parameters:

fn (Any) – a job function.

Returns:

the parameters model fn was registered with by add_parameters(), or None if it wasn’t.

Return type:

type[BaseModel] | None

Raises:

None. –

Example

from pyantz.infrastructure.config.parameters.decorators import (
    get_parameters,
)
from pyantz.jobs.files.simple import mkdir
from pyantz.jobs.testing.nop import do_nothing

print(get_parameters(mkdir).__name__, get_parameters(do_nothing))

Mkdirparams None

get_registered_functions(fn_name: str | None = None)

Get the functions which have used the add_parameters decorator.

A new list is returned each time, so callers can’t change the registry.

Parameters:

fn_name (str | None) – if given, only the functions with this full dotted name, or whose name ends with it at a . boundary (so copy finds x.simple.copy and y.moving.copy, but py finds nothing). All matches are returned.

Returns:

the matching registered functions.

Return type:

list[JobFunctionType]

Raises:

None. –

Example

import pyantz.jobs
from pyantz.infrastructure.config.parameters.decorators import (
    get_registered_functions,
)

names = sorted(fn.PYANTZ_NAME for fn in get_registered_functions("copy"))
print(names)

['pyantz.jobs.files.moving.copy', 'pyantz.jobs.files.simple.copy']

is_virtual(fn: Any)

Return true if the function is marked as virtual.

Parameters:

fn (Any) – the item to check.

Returns:

True if fn was marked by mark_virtual().

Return type:

bool

Raises:

None. –

Example

from pyantz.infrastructure.config import is_virtual
from pyantz.virtual.add_variables import AddVariables
from pyantz.jobs.testing.nop import do_nothing

print(is_virtual(AddVariables), is_virtual(do_nothing))

True False

mark_virtual[T](fn: T)

Add a parameter marking the function as a virtual job.

Used as a decorator on the parameters model of a virtual job, which must have a compile_virtual method (see VirtualParamModel).

Parameters:

fn (T) – the parameters model (or function) to mark.

Returns:

fn itself, marked.

Return type:

T

Raises:

AttributeError – if fn doesn’t accept new attributes.

Example

from pydantic import BaseModel

from pyantz.infrastructure.config import is_virtual, mark_virtual

@mark_virtual
class Nothing(BaseModel):
    def compile_virtual(self, deps):
        return deps

print(is_virtual(Nothing))

True

no_submit_fn[T: BaseModel | ParametersType](fn: Callable[[T], bool])

Wrap a function that doesn’t want submit function.

Every job is called with its parameters and a submit function; this lets a job which never submits other jobs take only its parameters. Apply it below add_parameters().

Parameters:

fn (Callable[[T], bool]) – the job function, taking only its parameters.

Returns:

a function taking the parameters and a submit function, which it ignores.

Return type:

Callable[[T, SubmissionFnType], bool]

Raises:

None. –

Example

from pyantz.infrastructure.config import no_submit_fn

@no_submit_fn
def check(params):
    return params["value"] > 0

print(check({"value": 1}, lambda job: None))

True

trace_fn[**P, T](fn: Callable[P, T])

Log when a function starts and finishes.

The messages are logged at debug level, to a logger named after the function.

Parameters:

fn (Callable[P, T]) – the function to wrap.

Returns:

the function, logging each call.

Return type:

Callable[P, T]

Raises:

None. –

Example

from pyantz.infrastructure.config.parameters.decorators import trace_fn

@trace_fn
def add(a, b):
    return a + b

print(add(1, 2))

3

update_deps[T: BaseModel](fn: Callable[[T, list[JobConfig]], list[JobConfig]])

Wrap a compilation function in a wrapper.

This will “scramble” the ids of all the jobs to reduce conflicts of multiple ids as they’re wrapped into virtual jobs. While other jobs should handle this, this will make life easier.

Parameters:

fn (Callable[[T, list[JobConfig]], list[JobConfig]]) – the compile_virtual method of a virtual job’s parameters model.

Returns:

the method, given the dependent jobs with new ids.

Return type:

Callable[[T, list[JobConfig]], list[JobConfig]]

Raises:

None. –

Example

from pyantz.infrastructure.config import JobConfig, update_deps

@update_deps
def compile_virtual(params, deps):
    return deps

nop = "pyantz.jobs.testing.nop.do_nothing"
first = JobConfig(job_id="first", function=nop, parameters={})
second = JobConfig(depends_on={"first"}, function=nop, parameters={})
new_first, new_second = compile_virtual(None, [first, second])
print(new_first.job_id != "first")
print(new_second.depends_on == {new_first.job_id})

True
True

Module contents

Functionality to mark functions to strongly type the parameters.