httk.workflow.languages.pwd¶

Running one Python Workflow Definition document as one httk₂ job.

The Python Workflow Definition (PWD) is a small JSON exchange format: a list of nodes and a list of edges. A node is a Python function named module.function, a literal input, or a named output; an edge connects one node’s output port to another node’s input port. The format is deliberately machine-facing — several workflow engines read and write it — and this module is httk₂ reading it.

from httk.workflow import Workspace, new_job

workspace = Workspace.initialize("workflow-workspace")
job = new_job(workspace, "workflow.json", parameters={"pwd_module_path": ["."]})

The import is one way and produces exactly one job. The whole graph runs inside that job, sequentially, in topological order, by the packaged pwd_runner.py — no runner file is written per workflow, and no per-node job is created: a PWD node is one Python call, which is not worth a claim, a lease and a process of its own.

The document travels in the job’s parameters when it fits within maximum_embedded_bytes, and in files/pwd.json with a document pointer when it does not, because parameters is bounded by MAXIMUM_PARAMETERS_BYTES and a generated document can be much larger than that.

Warning

Running a PWD document executes the Python functions it names. There is no sandbox and there cannot be one: the format’s whole content is module.function references. Import a document exactly as carefully as you would run the module it names. Passing allowed_modules records an allowlist of module prefixes in the job, which the runner refuses to import outside of.

Attributes¶

`MAXIMUM_PARAMETERS_BYTES`
`FILES_DIRECTORY`
`PACKAGE`
`RUNNER`
`KNOWN_VERSIONS`
`DOCUMENT_FILE`
`DEFAULT_MAXIMUM_EMBEDDED_BYTES`
`LANGUAGE`

Exceptions¶

PwdFormatError

A document that is not a Python Workflow Definition this importer accepts.

Classes¶

`LanguagePorts`	The named input and output ports of one language document.
`LanguageRequest`	The data supplied when preparing one language workflow.
`LanguageScaffold`	The files, runner, and hooks prepared for one language workflow.
`WorkflowLanguage`	The operations a workflow language exposes to the common layer.
`PwdDocument`	One validated PWD document, and the order its function nodes run in.

Functions¶

`runner_reference`(package, name)	Return the `runner` member of a `job.json` running one packaged runner.
`load_pwd_document`(path, *[, allow_unknown_version])	Read and validate one PWD document from path.
`validate_pwd_document`(raw, *[, source, ...])	Validate the shape of one PWD document and order its nodes.
`collect`(record)	Convert one PWD runner output document into provenance-capable records.

Package Contents¶

class httk.workflow.languages.pwd.LanguagePorts[source]¶

The named input and output ports of one language document.

Parameters:

inputs – Names of the document’s input ports.
outputs – Names of the document’s output ports.

inputs: tuple[str, Ellipsis]¶

outputs: tuple[str, Ellipsis]¶

class httk.workflow.languages.pwd.LanguageRequest[source]¶

The data supplied when preparing one language workflow.

Parameters:

workflow_id – Identify the workflow being prepared.
directory – Locate the workflow package, when it has one.
document – Locate the source workflow document, when it has one.
runner_options – Supply options for the language runner.
inputs – Describe the workflow inputs.
outputs – Describe the requested workflow outputs.
parameters – Describe the declared workflow parameters.
environment – Describe the declared workflow environment.
excluded_members – Package members the realization must not stage.

workflow_id: str¶

directory: pathlib.Path | None¶

document: pathlib.Path | None¶

runner_options: collections.abc.Mapping[str, object]¶

inputs: collections.abc.Mapping[str, collections.abc.Mapping[str, object]]¶

outputs: collections.abc.Mapping[str, collections.abc.Mapping[str, object]]¶

parameters: collections.abc.Mapping[str, collections.abc.Mapping[str, object]]¶

environment: collections.abc.Mapping[str, collections.abc.Mapping[str, object]]¶

excluded_members: tuple[str, Ellipsis] = ()¶

class httk.workflow.languages.pwd.LanguageScaffold[source]¶

The files, runner, and hooks prepared for one language workflow.

Parameters:

documents – Text or byte documents to write into the payload.
files – Files to stage into the payload.
parameters – Job parameters produced by preparation.
runner – Runner description for the job, when one is supplied.
runner_executor – Select the runner executor.
payload_runner – Name a runner staged in the payload.
workdir_path – Name the workdir below the job payload.
required_capabilities – Require these manager capabilities.
reserved_parameters – Names reserved for per-job realization output.
warnings – Preserve preparation warnings.
instantiate – Supply the per-job hook called after input staging.
finalize – Transform the per-job JobSpec immediately before its payload is prepared.

documents: collections.abc.Mapping[str, str | bytes]¶

files: collections.abc.Mapping[str, pathlib.Path]¶

parameters: collections.abc.Mapping[str, object]¶

runner: collections.abc.Mapping[str, object] | None¶

runner_executor: str = 'path'¶

payload_runner: str | None = None¶

workdir_path: str | None = None¶

required_capabilities: tuple[str, Ellipsis] = ()¶

reserved_parameters: tuple[str, Ellipsis] = ()¶

warnings: tuple[str, Ellipsis] = ()¶

instantiate: collections.abc.Callable[[httk.workflow.scaffold.InstantiateContext], object] | None = None¶

finalize: collections.abc.Callable[[httk.workflow.runtime_builders.JobSpec], httk.workflow.runtime_builders.JobSpec] | None = None¶

class httk.workflow.languages.pwd.WorkflowLanguage[source]¶

The operations a workflow language exposes to the common layer.

Parameters:

name – Name the language.
steps – Declare the runner’s steps.
initial_step – Select the runner’s initial step.
matches – Identify documents belonging to the language.
ports – Read the input and output ports of a document.
validate_runner – Validate runner options for a document.
prepare – Prepare a language request for execution.
collect – Convert a completed job record into language outputs.
document_policy – State whether package manifests require, allow, or forbid a source document.
open_ports – Skip manifest port validation when document ports cannot be enumerated statically.
has_default_collector – Provide a default collector path.
allows_modes – Permit manifest data and workdir mode overrides.
environment – Declare language-provided environment metadata.
required_modules – Name the importable modules a job of this language needs at run time.

name: str¶

steps: tuple[str, Ellipsis]¶

initial_step: str¶

matches: collections.abc.Callable[[pathlib.Path], bool]¶

ports: collections.abc.Callable[[pathlib.Path], LanguagePorts]¶

validate_runner: collections.abc.Callable[[collections.abc.Mapping[str, object], pathlib.Path], None]¶

prepare: collections.abc.Callable[[LanguageRequest], LanguageScaffold]¶

collect: collections.abc.Callable[[httk.workflow.collecting.JobRecord], collections.abc.Mapping[str, object]]¶

document_policy: DocumentPolicy = 'required'¶

open_ports: bool = False¶

has_default_collector: bool = True¶

allows_modes: bool = True¶

environment: collections.abc.Mapping[str, collections.abc.Mapping[str, object]]¶

required_modules: tuple[str, Ellipsis] = ()¶

httk.workflow.languages.pwd.runner_reference(package, name)[source]¶

Return the runner member of a job.json running one packaged runner.

The reserved pkg: form names the runner inside its own consumer package, and the digest is taken from the installed bytes, which is exactly what the manager verifies before it stages and executes them.

httk.workflow.languages.pwd.MAXIMUM_PARAMETERS_BYTES = 262144¶

httk.workflow.languages.pwd.FILES_DIRECTORY = 'files'[source]¶

httk.workflow.languages.pwd.PACKAGE = 'httk.workflow.languages.pwd'[source]¶

httk.workflow.languages.pwd.RUNNER = 'pwd_runner.py'[source]¶

httk.workflow.languages.pwd.KNOWN_VERSIONS = ('0.1.0',)[source]¶

httk.workflow.languages.pwd.DOCUMENT_FILE = 'files/pwd.json'[source]¶

httk.workflow.languages.pwd.DEFAULT_MAXIMUM_EMBEDDED_BYTES = 131072[source]¶

exception httk.workflow.languages.pwd.PwdFormatError[source]¶

Bases: ValueError

A document that is not a Python Workflow Definition this importer accepts.

class httk.workflow.languages.pwd.PwdDocument[source]¶

One validated PWD document, and the order its function nodes run in.

Parameters:

raw – Preserve the validated source document.
nodes – Preserve validated nodes keyed by identifier.
edges – Preserve validated graph edges.
order – Record the topological node execution order.

raw: collections.abc.Mapping[str, object][source]¶

nodes: collections.abc.Mapping[int, collections.abc.Mapping[str, object]][source]¶

edges: tuple[collections.abc.Mapping[str, object], Ellipsis][source]¶

order: tuple[int, Ellipsis][source]¶

property version: str | None[source]¶

Return the document version when one is declared.

property functions: tuple[str, Ellipsis][source]¶

Every module.function this document would import, in node order.

property input_names: tuple[str, Ellipsis][source]¶

The names of every input node, in node order.

property output_names: tuple[str, Ellipsis][source]¶

The names of every output node, in node order.

httk.workflow.languages.pwd.load_pwd_document(path, *, allow_unknown_version=False)[source]¶

Read and validate one PWD document from path.

Parameters:

path (str | os.PathLike[str]) – Read the PWD JSON document at this path.
allow_unknown_version (bool) – Try versions outside the supported set.

Returns:

The validated PWD document.

Raises:

httk.workflow.languages.pwd.PwdFormatError – If the file cannot be read, parsed, or validated.

Return type:

PwdDocument

httk.workflow.languages.pwd.validate_pwd_document(raw, *, source='the document', allow_unknown_version=False)[source]¶

Validate the shape of one PWD document and order its nodes.

Every member the format defines is checked; every member it does not define is preserved untouched, so a document carrying an engine’s own annotations survives the round trip into the job payload. When the python-workflow-definition package happens to be installed it is asked for a second opinion — it is never a dependency of httk-workflow, only a stricter validator when it is there.

Parameters:

raw (object) – Validate this decoded PWD document.
source (str) – Identify the document in validation errors.
allow_unknown_version (bool) – Try versions outside the supported set.

Returns:

The validated PWD document.

Raises:

httk.workflow.languages.pwd.PwdFormatError – If the document shape, graph, or version is invalid.

Return type:

PwdDocument

httk.workflow.languages.pwd.collect(record)[source]¶

Convert one PWD runner output document into provenance-capable records.

httk.workflow.languages.pwd.LANGUAGE[source]¶