httk.workflow.scaffold

Scaffolding submitted jobs from a workflow, some files, and some inputs.

A job is a payload directory plus a job.json that names the runner to execute, and building one by hand means knowing the runner’s workflow name, its initial step, its digest, and where its inputs live in the payload. This module is the short way: new_job() takes a workflow — a packaged runner a domain registered by name, or the path of a runner file of your own — stages the files the runner reads, writes the job.json, and submits the result, all in one call.

Packaged workflows are not known to this module. A domain or compat engine registers each one it ships with register_workflow(), supplying only the generic description of a starting point — its name, the runner it starts from, the workflow and steps that runner declares, the modes a job of it defaults to, and what it does — so the scaffold resolves and pins a workflow without ever importing the science that owns it.

from httk.workflow import Workspace
from httk.workflow.scaffold import new_job

workspace = Workspace.initialize("workflow-workspace")
job = new_job(workspace, "some-workflow", files={"input": "input"}, tag="example")
print(job.job_key, job.payload)

By default the runner file is published into the workspace runner store, and the job references it there by digest. That is what makes a scaffolded job durable: the bytes that will run are pinned in the workspace, so upgrading the installed httk-workflow underneath a queued campaign cannot change what its jobs execute. Publication is content addressed — the store name carries the digest of the bytes — so scaffolding the same workflow twice publishes nothing the second time, and a later, different version of a packaged runner lands beside the old one instead of replacing it. publish="installed" instead references a packaged workflow through the reserved pkg: form, which copies nothing at all.

new_jobs() is the same operation for a campaign: one workflow resolution and one publication amortized over every job, and a lazy iterator over the results. By design, generating a partitioned campaign costs one payload and one marker per job and never materializes a list of them.

Attributes

JOB_SCAFFOLD_FORMAT
DEFAULT_PLACEMENT
FILES_DIRECTORY
STRUCTURE_PATTERNS

Classes

WorkflowProvider	One packaged workflow a domain or compat engine offers by name.
JobItem	What one job of a new_jobs() campaign varies from the shared values.
ResolvedWorkflow	One resolved starting point for a job: a runner file and how to run it.
ScaffoldedJob	Describe one job this module submitted.
InstantiateContext	The mutable creation-time view passed to a runner's instantiate hook.

Functions

register_workflow(provider)	Register one packaged workflow, replacing any registered under its name.
registered_workflows()	Return registered ids, followed by sorted installed-plugin workflow ids.
registered_workflow_labels()	Return display labels for registered and installed-plugin workflows.
workflow_provider(name)	Return the provider selected by canonical id or alias.
describe_runner(runner)	Return the self-description one runner file prints, by running it.
registered_workflow(name)	Return the registered workflow selected by name, or None.
resolve_workflow(workflow, *[, workflow_id, step, ...])	Return the ResolvedWorkflow workflow names.
structure_files(directory)	Return every structure file of one directory, in a stable order.
structure_tag(path)	Return the job tag one structure file name suggests, or None.
new_job(workspace, workflow, *[, inputs, files, ...])	Scaffold, submit, and describe one job of workflow.
new_jobs(workspace, workflow, items, *[, inputs, ...])	Scaffold and submit one job per member of items, lazily.
payload_relative(name)	Return where one staged file lands inside a payload.

Module Contents

httk.workflow.scaffold.JOB_SCAFFOLD_FORMAT = 'httk-workflow-job-scaffold'

httk.workflow.scaffold.DEFAULT_PLACEMENT = 'jobs'

httk.workflow.scaffold.FILES_DIRECTORY = 'files'

httk.workflow.scaffold.STRUCTURE_PATTERNS = ('POSCAR*', '*.vasp')

class httk.workflow.scaffold.WorkflowProvider

One packaged workflow a domain or compat engine offers by name.

A provider is the generic description of a starting point. It names the packaged runner it starts from by the package the runner file is a module of and the file beside that module — so the reserved pkg: form and the digest are resolved from the provider alone — and it declares the runner’s workflow, its steps, the modes a job of it defaults to, and what it does. Declaring the steps here rather than running the runner to ask keeps scaffolding cheap; the owning domain’s tests hold the declaration to what the runner really describes.

Parameters:

• workflow_id – Name the workflow in registrations and job definitions.

• runner_package – Name the package containing a packaged runner.

• runner_file – Name the runner file beside the package module.

• language – Name the language runner realization, when applicable.

• document – Name the language document package member.

• runner_options – Supply language-specific runner options.

• initial_step – Select the default starting step.

• alias – Provide an alternate registered name.

• steps – Declare the steps the runner provides.

• data_mode – Declare the workflow’s default data mode.

• workdir_mode – Declare the workflow’s default workdir mode.

• summary – Describe the workflow for callers.

• inputs – Map input names to payload destinations or hook handling.

• instantiate – Indicate that the workflow has an instantiate hook.

• declarations – Declare workflow declaration documents.

• collector – Identify the optional result collector.

• directory – Locate a directory-sourced workflow package.

• build – Describe the package build command and generated artifacts.

• entry – Name the directory package’s runner entry.

• instantiate_file – Name the directory package’s instantiate hook.

• instantiate_exec – Name an executable directory package instantiate hook.

• collect_file – Name the directory package’s collector.

• collector_exec – Name an executable directory package collector.

• postprocess_scripts – Map curated postprocess script names to package members and descriptions.

• parameters – Declare the workflow’s parameter metadata.

• environment – Declare the workflow’s environment metadata.

• outputs – Declare the workflow’s output metadata.

• declaration_uri – Identify the source declaration URI.

• declaration_file – Name the source declaration file.

workflow_id: str

runner_package: str | None = None

runner_file: str | None = None

language: str | None = None

document: str | None = None

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

initial_step: str = 'start'

alias: str | None = None

steps: tuple[str, Ellipsis] = ()

data_mode: DataMode = 'none'

workdir_mode: WorkdirMode = 'persistent'

summary: str = ''

inputs: collections.abc.Mapping[str, str | None]

instantiate: bool = False

declarations: collections.abc.Mapping[str, collections.abc.Mapping[str, object]]

collector: collections.abc.Callable[[httk.workflow.collecting.JobRecord], collections.abc.Mapping[str, object]] | str | None = None

directory: pathlib.Path | None = None

build: httk.core.building.BuildSpec | None = None

entry: str = 'run'

instantiate_file: str | None = None

instantiate_exec: str | None = None

collect_file: str | None = None

collector_exec: str | None = None

postprocess_scripts: 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]]

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

declaration_uri: str | None = None

declaration_file: str | None = None

httk.workflow.scaffold.register_workflow(provider)

Register one packaged workflow, replacing any registered under its name.

A domain calls this once per workflow it ships when its package is imported, which is how httk workflow job new --workflow NAME resolves a packaged runner the generic scaffold never names.

Parameters:

provider (WorkflowProvider) – Supply the workflow provider to register.

Raises:

ValueError – If the provider’s id or alias collides with another registration.

httk.workflow.scaffold.registered_workflows()

Return registered ids, followed by sorted installed-plugin workflow ids.

Returns:

The registered workflow ids.

Return type:

tuple[str, Ellipsis]

httk.workflow.scaffold.registered_workflow_labels()

Return display labels for registered and installed-plugin workflows.

Returns:

The registered workflow labels, in registration order.

Return type:

tuple[str, Ellipsis]

httk.workflow.scaffold.workflow_provider(name)

Return the provider selected by canonical id or alias.

Parameters:

name (str) – Select a workflow by id or alias.

Returns:

The selected provider, or None when no provider matches.

Return type:

WorkflowProvider | None

class httk.workflow.scaffold.JobItem

Bases: TypedDict

What one job of a new_jobs() campaign varies from the shared values.

Every member is optional, and a member that is absent takes the value new_jobs() was called with. inputs and files are merged over the shared mappings key by key; everything else replaces the shared value.

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

files: collections.abc.Mapping[str, str | os.PathLike[str]]

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

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

tag: str | None

name: str

placement: str | pathlib.PurePosixPath

priority: int | None

class httk.workflow.scaffold.ResolvedWorkflow

One resolved starting point for a job: a runner file and how to run it.

A workflow is either one of the runners a domain registered by name — see registered_workflows() — or a runner file of your own, which is described by running it — every native runner answers --describe with its workflow and its steps — so a scaffolded job never guesses either.

Parameters:

• source – Locate the runner file or workflow package directory.

• workflow_id – Name the resolved workflow.

• language – Name the language runner realization, when applicable.

• document – Name the language document package member.

• runner_options – Preserve language-specific runner options.

• document_path – Locate the absolute language document.

• initial_step – Select the step a job starts at.

• alias – Preserve the registered workflow alias.

• steps – Preserve the steps the runner provides.

• data_mode – Preserve the workflow data mode.

• workdir_mode – Preserve the workflow workdir mode.

• packaged – Preserve the packaged runner file name when applicable.

• runner_package – Preserve the package containing packaged members.

• registration_id – Preserve the registration id when applicable.

• summary – Describe the resolved workflow.

• inputs – Map input names to payload destinations or hook handling.

• instantiate – Indicate that the workflow has an instantiate hook.

• declarations – Preserve workflow declaration documents.

• collector – Preserve the optional result collector.

• directory – Locate a directory-sourced workflow package.

• build – Describe the package build command and generated artifacts.

• entry – Name the directory package’s runner entry.

• instantiate_file – Name the directory package’s instantiate hook.

• instantiate_exec – Name an executable directory package instantiate hook.

• collect_file – Name the directory package’s collector.

• collector_exec – Name an executable directory package collector.

• postprocess_scripts – Preserve curated postprocess script metadata.

• parameters – Preserve the workflow’s parameter metadata.

• environment – Preserve the workflow’s environment metadata.

• outputs – Preserve the workflow’s output metadata.

• declaration_uri – Identify the source declaration URI.

• declaration_file – Name the source declaration file.

source: pathlib.Path

workflow_id: str

initial_step: str

language: str | None = None

document: str | None = None

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

document_path: pathlib.Path | None = None

alias: str | None = None

steps: tuple[str, Ellipsis] = ()

data_mode: DataMode = 'none'

workdir_mode: WorkdirMode = 'persistent'

packaged: str | None = None

runner_package: str | None = None

registration_id: str | None = None

summary: str = ''

inputs: collections.abc.Mapping[str, str | None]

instantiate: bool = False

declarations: collections.abc.Mapping[str, collections.abc.Mapping[str, object]]

collector: collections.abc.Callable[[httk.workflow.collecting.JobRecord], collections.abc.Mapping[str, object]] | str | None = None

directory: pathlib.Path | None = None

build: httk.core.building.BuildSpec | None = None

entry: str = 'run'

instantiate_file: str | None = None

instantiate_exec: str | None = None

collect_file: str | None = None

collector_exec: str | None = None

postprocess_scripts: 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]]

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

declaration_uri: str | None = None

declaration_file: str | None = None

property store_name: str

Return the content-addressed name this workflow takes in a runner store.

The digest of the bytes is part of the name, so publishing is idempotent for identical bytes and never overwrites a name a submitted job pinned: an upgraded packaged runner is published beside the version its queued jobs still reference.

Returns:

The digest-pinned runner-store name.

Return type:

str

class httk.workflow.scaffold.ScaffoldedJob

Describe one job this module submitted.

Parameters:

• job_id – Identify the submitted job.

• job_key – Identify the job payload and state markers.

• tag – Preserve the optional job tag.

• placement – Locate the job within the workspace.

• payload – Locate the submitted payload.

• marker – Locate the submitted state marker.

• workflow – Name the workflow the job runs.

• initial_step – Name the step the job starts at.

• runner – Describe the pinned runner.

• warnings – Preserve the preparation warnings raised for this workflow.

job_id: str

job_key: str

tag: str | None

placement: pathlib.PurePosixPath

payload: pathlib.Path

marker: pathlib.Path

workflow: str

initial_step: str

runner: collections.abc.Mapping[str, object]

warnings: tuple[str, Ellipsis] = ()

as_mapping()

Return the machine-readable report of this job.

Returns:

The serialized job report.

Return type:

dict[str, object]

class httk.workflow.scaffold.InstantiateContext

The mutable creation-time view passed to a runner’s instantiate hook.

payload is the staging root, inputs contains every supplied staged input, parameters is the merged opaque knob mapping, and tag is the caller’s tag. The hook may write below payload and update parameters; use suggest_tag() to provide a tag without overriding one the caller supplied.

Parameters:

• payload – Locate the payload being staged.

• inputs – Provide the supplied workflow inputs.

• parameters – Provide the merged job parameters.

• tag – Preserve or suggest the job tag.

payload: pathlib.Path

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

parameters: dict[str, object]

tag: str | None

suggest_tag(tag)

Set a tag only when the caller did not supply one.

httk.workflow.scaffold.describe_runner(runner)

Return the self-description one runner file prints, by running it.

Every native runner — Python or Bash — answers HTTK_WORKFLOW_DESCRIBE=1 with its workflow name and its registered steps and exits without touching anything, which is how a runner nobody wrote a workflow for is still scaffolded without being told what it implements.

Parameters:

runner (str | os.PathLike[str]) – Locate the runner file to describe.

Returns:

The validated runner description.

Raises:

ValueError – If the runner is missing, cannot run, or emits an invalid description.

Return type:

dict[str, object]

httk.workflow.scaffold.registered_workflow(name)

Return the registered workflow selected by name, or None.

Parameters:

name (str) – Select a workflow by id or alias.

Returns:

The resolved workflow, or None when no provider matches.

Return type:

ResolvedWorkflow | None

httk.workflow.scaffold.resolve_workflow(workflow, *, workflow_id=None, step=None, data_mode=None, format=None)

Return the ResolvedWorkflow workflow names.

workflow is the name of a packaged workflow, the file name of a packaged runner, or the path of a runner file. A runner file is described by running it, so its workflow name and its steps come from the runner itself; workflow and step override what it said, and step is required when a runner registers several steps and none of them is start.

Parameters:

• workflow (str | os.PathLike[str]) – Select a registered workflow, package directory, or runner file.

• workflow_id (str | None) – Override the resolved workflow id.

• step (str | None) – Override the resolved initial step.

• data_mode (DataMode | None) – Override the resolved data mode.

• format (str | None) – Force a language for a bare document or directory.

Returns:

The resolved workflow description.

Raises:

ValueError – If the workflow cannot be found or its description is invalid.

Return type:

ResolvedWorkflow

httk.workflow.scaffold.structure_files(directory)

Return every structure file of one directory, in a stable order.

A structure is a file matching one of STRUCTURE_PATTERNS — the VASP conventions POSCAR, POSCAR.something, and something.vasp — which is what makes a directory of structures one campaign.

Parameters:

directory (str | os.PathLike[str]) – Locate the directory to scan.

Returns:

Matching regular structure files in stable order.

Raises:

ValueError – If directory is not a directory.

Return type:

list[pathlib.Path]

httk.workflow.scaffold.structure_tag(path)

Return the job tag one structure file name suggests, or None.

The tag is the part of the name that identifies the structure — Si2O of POSCAR.Si2O, fcc-al of fcc-al.vasp — reduced to the tag syntax the protocol allows. A name that says nothing beyond POSCAR suggests no tag.

Parameters:

path (str | os.PathLike[str]) – Name the structure file whose tag to derive.

Returns:

The sanitized suggested tag, or None when no tag is present.

Return type:

str | None

httk.workflow.scaffold.new_job(workspace, workflow, *, inputs=None, files=None, parameters=None, environment=None, tag=None, placement=DEFAULT_PLACEMENT, priority=None, workdir_mode='persistent', data_mode=None, publish='workspace', step=None, format=None, workflow_id=None, name=None)

Scaffold, submit, and describe one job of workflow.

workflow is a registered workflow name — see registered_workflows() — or the path of a runner file. files maps payload names to the files to stage there: a bare name lands in the payload’s FILES_DIRECTORY, which is where a packaged runner reads its inputs, and a name with a directory in it is used verbatim. inputs stages the workflow’s declared objects into the payload; parameters is the job’s opaque implementation mapping.

data_mode defaults to what the workflow needs — transactional for a workflow whose runner publishes collected results, and none for a runner that said nothing. publish workspace publishes the runner file into the workspace runner store and pins its digest; installed references a packaged runner through the reserved pkg: form instead and copies nothing. It is ignored for language workflows, whose realization chooses the runner itself.

Parameters:

• workspace (httk.workflow.workspace.Workspace) – Provide the workspace receiving the job.

• workflow (str | os.PathLike[str]) – Select the workflow or runner file.

• inputs (collections.abc.Mapping[str, object] | None) – Supply declared workflow inputs.

• files (collections.abc.Mapping[str, str | os.PathLike[str]] | None) – Map payload names to files to stage.

• parameters (collections.abc.Mapping[str, object] | None) – Supply opaque job parameters.

• environment (collections.abc.Mapping[str, object] | None) – Supply overrides for declared workflow environment values.

• tag (str | None) – Set the job tag.

• placement (str | pathlib.PurePosixPath) – Place the job within the workspace.

• priority (int | None) – Set the scheduling priority.

• workdir_mode (WorkdirMode) – Select the job workdir mode.

• data_mode (DataMode | None) – Override the workflow data mode.

• publish (PublishMode) – Select workspace publication or installed reference.

• step (str | None) – Override the workflow’s initial step.

• format (str | None) – Force a language for a bare workflow document or directory.

• workflow_id (str | None) – Override the workflow id in the job definition.

• name (str | None) – Set the job’s display name.

Returns:

The submitted job description.

Raises:

ValueError – If workflow, inputs, placement, or job settings are invalid.

Return type:

ScaffoldedJob

httk.workflow.scaffold.new_jobs(workspace, workflow, items, *, inputs=None, files=None, parameters=None, environment=None, tag=None, placement=DEFAULT_PLACEMENT, priority=None, workdir_mode='persistent', data_mode=None, publish='workspace', step=None, format=None, workflow_id=None, name=None)

Scaffold and submit one job per member of items, lazily.

Every keyword is the shared value of the whole campaign, and every member of one JobItem is what that job varies: inputs and files are merged over the shared mappings, and tag, name, placement, and priority replace the shared value.

This is the pattern for a campaign of any size. The workflow is resolved once and its runner published once, however many jobs follow, so every job costs exactly one payload directory and one state marker; items is consumed as an iterator and the results are yielded as they are submitted, so a structure generator can be turned into jobs without either side of the loop ever being materialized.

Parameters:

• workspace (httk.workflow.workspace.Workspace) – Provide the workspace receiving the jobs.

• workflow (str | os.PathLike[str]) – Select the workflow or runner file.

• items (collections.abc.Iterable[JobItem]) – Yield per-job overrides.

• inputs (collections.abc.Mapping[str, object] | None) – Supply shared declared workflow inputs.

• files (collections.abc.Mapping[str, str | os.PathLike[str]] | None) – Supply shared payload files.

• parameters (collections.abc.Mapping[str, object] | None) – Supply shared opaque job parameters.

• environment (collections.abc.Mapping[str, object] | None) – Supply shared declared environment overrides.

• tag (str | None) – Set the shared job tag.

• placement (str | pathlib.PurePosixPath) – Set the shared workspace placement.

• priority (int | None) – Set the shared scheduling priority.

• workdir_mode (WorkdirMode) – Select the shared workdir mode.

• data_mode (DataMode | None) – Override the workflow data mode.

• publish (PublishMode) – Select workspace publication or installed reference.

• step (str | None) – Override the workflow’s initial step.

• format (str | None) – Force a language for a bare workflow document or directory.

• workflow_id (str | None) – Override the workflow id in each job definition.

• name (str | None) – Set the shared display name.

Returns:

An iterator yielding each submitted job description.

Yield:

Each submitted job description.

Raises:

ValueError – If workflow, inputs, placement, or job settings are invalid.

Return type:

collections.abc.Iterator[ScaffoldedJob]

def structures():
    for path in sorted(Path("structures").glob("POSCAR.*")):
        yield {"files": {"POSCAR": path}, "tag": structure_tag(path)}

for job in new_jobs(workspace, "some-workflow", structures(), parameters={"kpoint_density": 30.0}):
    print(job.job_key)

httk.workflow.scaffold.payload_relative(name)

Return where one staged file lands inside a payload.

A bare name lands in FILES_DIRECTORY, so POSCAR becomes files/POSCAR — where the packaged runners read it — and a name that carries a directory of its own is used exactly as it is written.

Parameters:

name (str) – Name the staged file within the payload.

Returns:

The validated payload-relative destination.

Raises:

• httk.workflow.errors.FormatError – If the name uses a reserved payload member.

• ValueError – If the name is empty or absolute.

Return type:

pathlib.PurePosixPath