pyantz package

pyantz package#

Subpackages#

Submodules#

pyantz.__main__ module#

Primary way to run a configuration.

Pass a path to the configuration file (JSON or YAML):

pyantz config.json
pyantz config.yaml --validate
pyantz config.json --log-level DEBUG --log-file run.log

The exit status is 1 if the configuration is invalid or a job failed.

LOG_LEVELS = ('DEBUG', 'INFO', 'WARNING', 'ERROR')#

Choices of --log-level.

YAML_SUFFIXES = ('.yaml', '.yml')#

Endings of configuration files read as YAML; others are read as JSON.

_build_parser()[source]#

Create the command line parser.

Parameters:

None.

Returns:

the parser of pyantz’s arguments.

Return type:

ArgumentParser

Raises:

None. –

Example

from pyantz.__main__ import _build_parser

args = _build_parser().parse_args(["config.yaml", "--log-level", "debug"])
print(args.config_location, args.log_level, args.validate)

config.yaml DEBUG False
_describe(config: InitialConfig)[source]#

List the jobs of the configuration and what they depend on.

Parameters:

config (InitialConfig) – the validated configuration.

Returns:

one line per job: its id, name and the ids it depends on.

Return type:

str

Raises:

None. –

Example

from pyantz.__main__ import _describe
from pyantz.infrastructure.config import InitialConfig

nop = "pyantz.jobs.testing.nop.do_nothing"
config = InitialConfig.model_validate(
    {
        "submitter": {"type_": "local_proc", "working_directory": "."},
        "jobs": [
            {"job_id": "a", "name": "first", "function": nop,
             "parameters": {}},
            {"job_id": "b", "function": nop, "parameters": {},
             "depends_on": ["a"]},
        ],
    }
)
print(_describe(config))

2 job(s):
  a (first)
  b <- a
_load_config_data(path: Path)[source]#

Read a JSON or YAML file into plain python values.

Parameters:

path (Path) – the file; read as YAML if it ends with .yaml or .yml.

Returns:

the file’s contents.

Return type:

Any

Raises:

  • OSError – if the file can’t be read.

  • ValueError – if a JSON file is invalid.

  • ruamel.yaml.error.YAMLError – if a YAML file is invalid.

Example

from pathlib import Path

from pyantz.__main__ import _load_config_data

Path("config.yaml").write_text("jobs: []\nvariables: {name: test}\n")
print(_load_config_data(Path("config.yaml")))

{'jobs': [], 'variables': {'name': 'test'}}
_setup_logging(level: str, log_file: str | None)[source]#

Show the logs of pyantz at the requested level.

Replaces any logging setup already made.

Parameters:

  • level (str) – the lowest level shown, e.g., "INFO".

  • log_file (str | None) – if set, also write the log to this file.

Return type:

None

Returns:

None.

Raises:

OSError – if the log file can’t be opened.

main()[source]#

Entry point for running pyantz.

Reads the command line (see _build_parser()), checks the configuration and runs it, or only describes it with --validate.

Parameters:

None.

Return type:

None

Returns:

None.

Raises:

SystemExit – with status 1 if the configuration is invalid or a job failed, 2 if the arguments are invalid or the file doesn’t exist.

pyantz.status module#

Show the progress of a run: pyantz-status <working directory>.

QUEUE_FILE_NAME = 'queue.db'#

Name of the queue file the local runner keeps in its working directory.

describe_run(working_directory: Path)[source]#

Summarize the queue of the run in working_directory.

Parameters:

working_directory (Path) – the working directory of a local run.

Returns:

the number of jobs in each status, and the ids of the failed jobs.

Return type:

str

Raises:

FileNotFoundError – if the directory has no queue (no run was started there)

Example

from pathlib import Path

from pyantz import start
from pyantz.status import describe_run

start(
    {
        "submitter": {"type_": "local_proc", "working_directory": "run"},
        "jobs": [
            {
                "job_id": "check",
                "function": "pyantz.jobs.files.check.file_exists",
                "parameters": {"file": "missing.txt"},
            },
        ],
    }
)
print(describe_run(Path("run")))

Jobs: 1
  succeeded: 0
  failed:    1
  running:   0
  ready:     0
  waiting:   0
Failed jobs:
  check
main()[source]#

Entry point of pyantz-status.

Parameters:

None.

Return type:

None

Returns:

None.

Raises:

SystemExit – with status 1 if the directory has no run, 2 if the arguments are invalid.

Module contents#

PyAntz is a simple local DAGster/Airflow-style data pipeline tool.

It is useful if you want to quickly run some small things locally or on a slurm cluster without setting up specialized worker nodes.