pyantz.api.endpoints.pipelines package

pyantz.api.endpoints.pipelines package#

Submodules#

pyantz.api.endpoints.pipelines.endpoints module#

Endpoints to validate, preview and run pipeline configurations.

type ConfigBody = Annotated[dict[str, Any], Body(PydanticUndefined)]#

A configuration, sent as the request’s json body.

DEFAULT_LOG_BYTES = 20000#

How much of a run’s log a request gets by default.

MAX_LOG_BYTES = 1000000#

The most of a run’s log which one request can get.

RUN_ID_PATTERN = re.compile('[0-9a-f]{32}')#

What a run id looks like (a uuid in hex).

_PROCESSES: dict[str, Popen[bytes]] = {}#

The processes started by this server, by run id; a restart forgets them.

_problems(config: dict[str, Any])[source]#

Validate the configuration and list what is wrong with it.

Parameters:

config (dict[str, Any]) – the configuration, as sent.

Returns:

the validated configuration (None if invalid) and a message for each problem.

Return type:

tuple[InitialConfig | None, list[str]]

Raises:

None – problems are returned.

Example

from pyantz.api.endpoints.pipelines.endpoints import _problems

config, problems = _problems({"jobs": []})
print(config, problems)

None ['submitter: Field required']
_run_dir(run_id: str)[source]#

Get the directory of a run, refusing ids which could escape the runs dir.

Parameters:

run_id (str) – the run’s id.

Returns:

the run’s directory (which may not exist).

Return type:

Path

Raises:

HTTPException – 404 if the id isn’t a valid run id.

Example

from fastapi import HTTPException

from pyantz.api.endpoints.pipelines.endpoints import _run_dir

try:
    _run_dir("../../etc")
except HTTPException as exc:
    print(exc.status_code)

404
compile_config(config: ConfigBody)[source]#

Compile a configuration into its concrete jobs and dependency edges.

Parameters:

config (ConfigBody) – the configuration, as sent.

Returns:

the concrete jobs (virtual jobs expanded) and the edges between them.

Return type:

CompileResponse

Raises:

HTTPException – 422 if the configuration is invalid.

Example

from pyantz.api.endpoints.pipelines.endpoints import compile_config

nop = "pyantz.jobs.testing.nop.do_nothing"
result = compile_config(
    {
        "submitter": {"type_": "local_proc", "working_directory": "."},
        "jobs": [
            {"job_id": "a", "function": nop, "parameters": {}},
            {"job_id": "b", "function": nop, "parameters": {},
             "depends_on": ["a"]},
        ],
    }
)
print(result.edges)

[('a', 'b')]
get_run(run_id: str)[source]#

Get the progress of a run: its state, job counts and per-job statuses.

Parameters:

run_id (str) – the run’s id.

Returns:

the run’s state, the counts of its jobs and their statuses.

Return type:

RunResponse

Raises:

HTTPException – 404 if there is no such run.

Example

import os
from pathlib import Path

from pyantz.api.endpoints.pipelines.endpoints import get_run

os.environ["PYANTZ_RUNS_DIR"] = "runs"
run_id = "0" * 32
Path("runs", run_id).mkdir(parents=True)
try:
    print(get_run(run_id).state)  # this server didn't start it
finally:
    del os.environ["PYANTZ_RUNS_DIR"]

unknown
get_run_log(run_id: str, max_bytes: int = 20000)[source]#

Get the end of what a run printed (its standard output and errors).

Parameters:

  • run_id (str) – the run’s id.

  • max_bytes (int) – how many bytes of the end of the log to return.

Returns:

the end of the log, and whether it was cut.

Return type:

RunLogResponse

Raises:

  • HTTPException – 404 if the run has no log.

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

Example

import os
from pathlib import Path

from pyantz.api.endpoints.pipelines.endpoints import get_run_log

os.environ["PYANTZ_RUNS_DIR"] = "runs"
run_id = "0" * 32
Path("runs", run_id).mkdir(parents=True)
Path("runs", run_id, "run.log").write_text("starting... all done")
try:
    result = get_run_log(run_id, max_bytes=8)
    print(result.log, result.truncated)
finally:
    del os.environ["PYANTZ_RUNS_DIR"]

all done True
router = <fastapi.routing.APIRouter object>#

Routes checking configurations.

Parameters:

  • scope (Scope)

  • receive (Receive)

  • send (Send)

Return type:

None

runs_router = <fastapi.routing.APIRouter object>#

Routes starting and following runs.

Parameters:

  • scope (Scope)

  • receive (Receive)

  • send (Send)

Return type:

None

start_run(config: ConfigBody)[source]#

Start a run of the configuration in a new process, in its own directory.

Only local runs are accepted, and the working directory is chosen by the server. Since a configuration can import and run any function, runs are refused unless the server limits the modules with PYANTZ_ALLOWED_MODULES.

The run’s directory (in get_runs_dir()) holds its configuration, its log (run.log) and its working directory (work).

Parameters:

config (ConfigBody) – the configuration, as sent.

Returns:

the id of the new run.

Return type:

StartRunResponse

Raises:

  • HTTPException – 403 if PYANTZ_ALLOWED_MODULES isn’t set, 422 if the configuration is invalid or isn’t a local run.

  • OSError – if the run’s directory can’t be created.

validate_config(config: ConfigBody)[source]#

Check a configuration and list its problems.

Parameters:

config (ConfigBody) – the configuration, as sent.

Returns:

whether it is valid, and its problems.

Return type:

ValidateResponse

Raises:

None – problems are returned.

Example

from pyantz.api.endpoints.pipelines.endpoints import validate_config

result = validate_config(
    {
        "submitter": {"type_": "local_proc", "working_directory": "."},
        "jobs": [
            {"function": "pyantz.jobs.testing.nop.do_nothing",
             "parameters": {}, "depends_on": ["missing"]},
        ],
    }
)
print(result.valid, len(result.errors))

False 1

pyantz.api.endpoints.pipelines.models module#

Data models of the pipeline endpoints.

class CompileResponse(*, jobs: list[CompiledJob], edges: list[tuple[str, str]])[source]#

Bases: BaseModel

The jobs of a configuration once virtual jobs are expanded.

Parameters:

  • jobs (list[CompiledJob])

  • edges (list[tuple[str, str]])

_abc_impl = <_abc._abc_data object>#
edges: list[tuple[str, str]]#

(dependency, dependent) pairs, ready to draw as a graph.

jobs: list[CompiledJob]#

The concrete jobs.

class CompiledJob(*, job_id: str, name: str | None = None, function: str, depends_on: list[str] = <factory>)[source]#

Bases: BaseModel

A job of the compiled configuration.

Parameters:

  • job_id (str)

  • name (str | None)

  • function (str)

  • depends_on (list[str])

_abc_impl = <_abc._abc_data object>#
depends_on: list[str]#

Ids of the jobs it depends on, sorted.

function: str#

Import path of the job’s function.

job_id: str#

The job’s id.

name: str | None#

The job’s human readable name.

class JobState(*, job_id: str, status: str, details: str | None = None)[source]#

Bases: BaseModel

Status of one job of a run.

Parameters:

  • job_id (str)

  • status (str)

  • details (str | None)

_abc_impl = <_abc._abc_data object>#
details: str | None#

Why the job has this status, e.g., its error.

job_id: str#

The job’s id.

status: str#

The job’s status, in lower case (ready, pending, …).

class RunLogResponse(*, run_id: str, log: str, truncated: bool)[source]#

Bases: BaseModel

The end of what a run printed.

Parameters:

  • run_id (str)

  • log (str)

  • truncated (bool)

_abc_impl = <_abc._abc_data object>#
log: str#

The last bytes of the log, decoded; may start in the middle of a line.

run_id: str#

The run’s id.

truncated: bool#

True if the log is longer than what is returned.

class RunResponse(*, run_id: str, state: str, report: dict[str, ~typing.Any] | None=None, jobs: list[JobState] = <factory>)[source]#

Bases: BaseModel

Progress of a run.

Parameters:

  • run_id (str)

  • state (str)

  • report (dict[str, Any] | None)

  • jobs (list[JobState])

_abc_impl = <_abc._abc_data object>#
jobs: list[JobState]#

The status of each job.

report: dict[str, Any] | None#

Counts of the jobs by status; None until the run has created its queue.

run_id: str#

The run’s id.

state: str#

running, succeeded, failed or unknown (e.g. the server restarted mid-run).

class StartRunResponse(*, run_id: str)[source]#

Bases: BaseModel

A run was started.

Parameters:

run_id (str)

_abc_impl = <_abc._abc_data object>#
run_id: str#

Id to ask about the run with.

class ValidateResponse(*, valid: bool, errors: list[str] = <factory>)[source]#

Bases: BaseModel

Result of checking a configuration.

Parameters:

  • valid (bool)

  • errors (list[str])

_abc_impl = <_abc._abc_data object>#
errors: list[str]#

One message per problem found, empty if valid.

valid: bool#

True if the configuration has no problems.

Module contents#

Validate, preview and run pipeline configurations.