Python API Reference¶

This reference documents the public Python surfaces that are intended for user workflows, tool authoring, compute payload construction, and compute submission.

For guided learning, start with Python Workflow API, Building Tool Contracts in Python, and From Python Workflow to Compute Payload. Use this page when you need signatures and member-level detail.

sophios.apis.python.workflow and sophios.apis.python.tool_builder¶

Import user-facing workflow and tool-authoring objects from their concrete modules:

from sophios.apis.python.workflow import Step, Workflow
from sophios.apis.python.tool_builder import CommandLineTool, Input, Output

The detailed member documentation lives in the concrete modules below.

sophios.apis.python.workflow¶

Python API for building Sophios workflows that compile to CWL.

class sophios.apis.python.workflow.DisableEverythingFilter(name='')¶

filter(record)¶

Determine if the specified record is to be logged.

Returns True if the record should be logged, or False otherwise. If deemed appropriate, the record may be modified in-place.

Return type:: bool

class sophios.apis.python.workflow.Step(clt_path, config_path=None, *, step_name=None, tool_registry=None)¶

A workflow step backed by a CWL CommandLineTool.

Attribute writes like step.message = "hi" bind named step inputs. Attribute reads like step.output_file resolve named step outputs. The same ports are also available through the explicit step.inputs.* and step.outputs.* namespaces.

clt: CommandLineTool | CommandLineTool | CommandLineTool¶

clt_path: Path¶

process_name: str¶

cwl_version: str¶

yaml: dict[str, Any]¶

cfg_yaml: dict[str, Any]¶

inputs: ParameterNamespace[InputParameter, InputParameter]¶

outputs: ParameterNamespace[OutputParameter, OutputParameter]¶

scatter: list[InputParameter]¶

scatterMethod: str¶

when: str¶

__init__(clt_path, config_path=None, *, step_name=None, tool_registry=None)¶

Create a Step from a CWL file or CommandLineTool-like object.

Parameters:

clt_path (Any) – Path to a CWL tool definition, or an object with name and to_dict() such as tool_builder.CommandLineTool.
config_path (StrPath | None) – Optional YAML config used to pre-bind file-backed step inputs.
step_name (str | None) – Optional workflow step name override.
tool_registry (Tools | None) – Optional fallback registry for known tools.

Raises:

TypeError – If the source or config uses an unsupported type.
InvalidCLTError – If the CWL tool cannot be loaded from disk or the registry.

Returns:

The step is initialized in place.

Return type:

None

classmethod from_cwl(document, *, process_name=None, run_path=None, config=None, tool_registry=None)¶

Create a Step from an in-memory CWL CommandLineTool document.

Parameters:

document (Mapping[str, Any]) – Parsed CWL CommandLineTool fields.
process_name (str | None) – Optional step name override.
run_path (StrPath | None) – Optional virtual .cwl path for compiler bookkeeping.
config (Mapping[str, Any] | None) – Optional input values to pre-bind.
tool_registry (Tools | None) – Optional tool registry retained on the step.

Raises:

TypeError – If run_path uses an unsupported type.
InvalidCLTError – If the CWL document cannot be parsed.

Returns:

A fully initialized step backed by the in-memory tool.

Return type:

Step

bind_input(name, value)¶

Bind a value or upstream output to a named step input parameter.

Parameters:

name (str) – The input parameter name.
value (Any) – A literal value, a workflow input reference, or a step output.

Raises:

AttributeError – If the named input does not exist on the step.

Returns:

The step is mutated in place.

Return type:

None

scatter_on(*inputs, method='dotproduct')¶

Scatter this step over one or more already-bound input parameters.

Parameters:

inputs (InputParameter) – Input parameters on this step to scatter over.
method (str | ScatterMethod) – CWL scatter method. Defaults to dotproduct.

Returns:

This step, so calls can be chained in small examples.

Return type:

Step

get_inp_attr(name)¶

Return a named input parameter from this step.

Parameters:: name (str) – The input parameter name.
Raises:: AttributeError – If the input does not exist.
Returns:: The requested step input parameter.
Return type:: InputParameter

get_output(name)¶

Return a named output parameter from this step.

Parameters:: name (str) – The output parameter name.
Raises:: AttributeError – If the output does not exist.
Returns:: The requested step output parameter.
Return type:: OutputParameter

flatten_steps()¶

Return this step as a single-item list for recursive traversal.

Return type:: list[Step]

flatten_subworkflows()¶

Return an empty subworkflow list because steps do not nest workflows.

Return type:: list[Workflow]

class sophios.apis.python.workflow.Workflow(steps, workflow_name)¶

A Sophios workflow composed from Step objects and nested Workflow objects.

steps: list[Step | Workflow]¶

process_name: str¶

inputs: ParameterNamespace[InputParameter, WorkflowInputReference]¶

outputs: ParameterNamespace[OutputParameter, OutputParameter]¶

yml_path: Optional[Path]¶

__init__(steps, workflow_name)¶

Create a workflow from steps and/or nested subworkflows.

Parameters:

steps (Sequence[Step | Workflow]) – Child workflow nodes in execution order.
workflow_name (str) – User-facing workflow name.

Returns:

The workflow is initialized in place.

Return type:

None

add_input(name, parameter_type=None)¶

Declare a workflow input explicitly.

Parameters:

name (str) – The workflow input name.
parameter_type (Any) – Optional CWL type expression for the input.

Returns:

The created or existing workflow input parameter.

Return type:

InputParameter

add_output(name, source=None, *, parameter_type=None, implicit=False)¶

Declare a workflow output explicitly.

Parameters:

name (str) – The workflow output name.
source (Any) – Optional step output or workflow input reference to expose.
parameter_type (Any) – Optional CWL type expression for the output.

Returns:

The created or existing workflow output parameter.

Return type:

OutputParameter

bind_input(name, value)¶

Bind a literal value or upstream output to a workflow input.

Parameters:

name (str) – The workflow input name.
value (Any) – A literal value, workflow reference, or step output.

Returns:

The workflow is mutated in place.

Return type:

None

bind_output(name, value)¶

Bind a named workflow output to a step output or workflow input.

Parameters:

name (str) – The workflow output name.
value (Any) – A step output or workflow input reference to expose.

Returns:

The workflow is mutated in place.

Return type:

None

get_inp_attr(name)¶

Return a named workflow input, creating it if needed.

Parameters:: name (str) – The workflow input name.
Returns:: The created or existing workflow input parameter.
Return type:: InputParameter

append(step_)¶

Append a step or nested workflow to this workflow.

Parameters:: step (Any) – The Step or Workflow to append.
Raises:: TypeError – If step_ is neither a Step nor a Workflow.
Returns:: The workflow is mutated in place.
Return type:: None

property yaml: dict[str, Any]¶

Return the in-memory WIC YAML representation of this workflow.

Returns:: A WIC-compatible YAML tree represented as a Python dict.
Return type:: dict[str, Any]

write_ast_to_disk(directory)¶

Write this workflow tree to disk as sibling .wic files.

This compatibility method is retained for existing callers. New code should prefer write_wic(), which can write either one inline document or a sibling-file tree.

Parameters:: directory (Path) – Directory where the workflow AST should be written.
Returns:: Files are written to disk as a side effect.
Return type:: None

to_wic(*, inline_subworkflows=True)¶

Return this workflow as .wic YAML text.

Parameters:: inline_subworkflows (bool) – Whether nested workflows should be embedded in the returned document.
Returns:: The serialized .wic document.
Return type:: str

write_wic(path=None, *, inline_subworkflows=True)¶

Write this workflow as a .wic file.

Parameters:

path (StrPath | None) – Destination .wic path or output directory. When omitted, writes <workflow>.wic in the current directory.
inline_subworkflows (bool) – Whether nested workflows should be embedded in the root file. When false, nested workflows are written as sibling .wic files beside the root document.

Returns:

The root .wic file that was written.

Return type:

Path

flatten_steps()¶

Return every concrete step in this workflow tree.

Returns:: All Step instances reachable from this workflow.
Return type:: list[Step]

flatten_subworkflows()¶

Return this workflow and all nested subworkflows.

Returns:: This workflow followed by nested subworkflows.
Return type:: list[Workflow]

compile(write_to_disk=False, *, tool_registry=None)¶

Compile this workflow into CWL.

Parameters:

write_to_disk (bool) – Whether to also write generated CWL to autogenerated/.
tool_registry (Tools | None) – Optional tool registry override.

Returns:

The compiler result tree for this workflow.

Return type:

CompilerInfo

write_artifacts(*, tool_registry=None)¶

Compile this workflow and write generated CWL artifacts to disk.

Parameters:: tool_registry (Tools | None) – Optional tool registry override.
Returns:: The compiler result tree for this workflow.
Return type:: CompilerInfo

get_cwl_workflow(*, tool_registry=None)¶

Return the compiled CWL workflow JSON and generated input object.

Parameters:: tool_registry (Tools | None) – Optional tool registry override.
Returns:: A JSON-serializable representation of the compiled CWL workflow.
Return type:: Json

run(run_args_dict=None, user_env_vars=None, basepath='autogenerated', tool_registry=None)¶

Compile and execute this workflow locally.

Parameters:

run_args_dict (dict[str, str] | None) – Runtime CLI options for local execution.
user_env_vars (dict[str, str] | None) – Environment variables to expose to the run.
basepath (str) – Directory used for generated files and execution artifacts.
tool_registry (Tools | None) – Optional tool registry override.

Returns:

The workflow is executed as a side effect.

Return type:

None

sophios.apis.python.tool_builder¶

Public CWL v1.2 CommandLineTool authoring API.

The required core is intentionally small:

`python inputs = Inputs(input=Input(cwl.directory, position=1)) outputs = Outputs(output=Output(cwl.directory, from_input=inputs.input)) tool = CommandLineTool("example", inputs, outputs) `

Everything else is optional and chainable.

exception sophios.apis.python.tool_builder.ToolBuilderValidationError¶: Raised when a generated CLT fails schema validation.

class sophios.apis.python.tool_builder.CommandArgument(value=None, binding=None, extra=<factory>)¶

A structured CWL command-line argument.

value: Any¶

binding: Optional[CommandLineBinding]¶

extra: dict[str, Any]¶

to_yaml()¶

Return type:: str | dict[str, Any]

__init__(value=None, binding=None, extra=<factory>)¶

class sophios.apis.python.tool_builder.CommandLineBinding(position=None, prefix=None, separate=None, item_separator=None, value_from=None, shell_quote=None, extra=<factory>)¶

A CWL input binding or argument binding.

position: UnionType[int, float, None]¶

prefix: Optional[str]¶

separate: Optional[bool]¶

item_separator: Optional[str]¶

value_from: Any¶

shell_quote: Optional[bool]¶

extra: dict[str, Any]¶

to_dict()¶

Return type:: dict[str, Any]

__init__(position=None, prefix=None, separate=None, item_separator=None, value_from=None, shell_quote=None, extra=<factory>)¶

class sophios.apis.python.tool_builder.CommandLineTool(name, inputs, outputs, cwl_version='v1.2', label_text=None, doc_text=None, _base_command=<factory>, _arguments=<factory>, _requirements=<factory>, _hints=<factory>, _stdin=None, _stdout=None, _stderr=None, _intent=<factory>, _namespaces=<factory>, _schemas=<factory>, _success_codes=<factory>, _temporary_fail_codes=<factory>, _permanent_fail_codes=<factory>, _extra=<factory>)¶

Declarative CWL CommandLineTool authoring object.

name: str¶

inputs: Inputs¶

outputs: Outputs¶

cwl_version: str¶

label_text: Optional[str]¶

doc_text: UnionType[str, list[str], None]¶

describe(label=None, doc=None)¶

Return type:: CommandLineTool

label(text)¶

Return type:: CommandLineTool

doc(text)¶

Return type:: CommandLineTool

namespace(prefix, iri=None)¶

Return type:: CommandLineTool

schema(iri)¶

Return type:: CommandLineTool

edam()¶

Return type:: CommandLineTool

intent(*identifiers)¶

Return type:: CommandLineTool

base_command(*parts)¶

Return type:: CommandLineTool

stdin(value)¶

Return type:: CommandLineTool

stdout(value)¶

Return type:: CommandLineTool

stderr(value)¶

Return type:: CommandLineTool

add_argument(argument)¶

Return type:: CommandLineTool

argument(value=None, **kwargs)¶

Return type:: CommandLineTool

requirement(requirement, value=None)¶

Return type:: CommandLineTool

hint(requirement, value=None)¶

Return type:: CommandLineTool

docker(image=None, *, as_hint=False, **kwargs)¶

Return type:: CommandLineTool

inline_javascript(*expression_lib, as_hint=False, extra=None)¶

Return type:: CommandLineTool

schema_definitions(*types, as_hint=False, extra=None)¶

Return type:: CommandLineTool

load_listing(value, *, as_hint=False, extra=None)¶

Return type:: CommandLineTool

shell_command(*, as_hint=False, extra=None)¶

Return type:: CommandLineTool

software(packages, *, as_hint=False, extra=None)¶

Return type:: CommandLineTool

initial_workdir(listing, *, as_hint=False, extra=None)¶

Return type:: CommandLineTool

stage(reference, *, writable=False, entryname=None, as_hint=False, extra=None)¶

Return type:: CommandLineTool

env_var(name, value, *, as_hint=False)¶

Return type:: CommandLineTool

resources(*, as_hint=False, extra=None, **kwargs)¶

Return type:: CommandLineTool

gpu(*, cuda_version_min=None, compute_capability=None, device_count_min=None, as_hint=True, extra=None)¶

Return type:: CommandLineTool

work_reuse(enable, *, as_hint=False, extra=None)¶

Return type:: CommandLineTool

network_access(enable, *, as_hint=False, extra=None)¶

Return type:: CommandLineTool

inplace_update(enable=True, *, as_hint=True, extra=None)¶

Return type:: CommandLineTool

time_limit(seconds, *, as_hint=False, extra=None)¶

Return type:: CommandLineTool

success_codes(*codes)¶

Return type:: CommandLineTool

temporary_fail_codes(*codes)¶

Return type:: CommandLineTool

permanent_fail_codes(*codes)¶

Return type:: CommandLineTool

extra(**values)¶

Return type:: CommandLineTool

to_step(*, step_name=None, run_path=None, config=None, tool_registry=None)¶

Convert this built CLT into an in-memory workflow Step.

Parameters:

step_name (str | None) – Optional workflow step name override.
run_path (str | Path | None) – Optional virtual .cwl path for compiler bookkeeping.
config (dict[str, Any] | None) – Optional input values to pre-bind.
tool_registry (Tools | None) – Optional tool registry retained on the step.

Returns:

A workflow step backed by this CLT without writing to disk.

Return type:

Step

build()¶

Return type:: dict[str, Any]

to_dict()¶

Return type:: dict[str, Any]

to_yaml()¶

Return type:: str

save(path, *, validate=False, skip_schemas=False)¶

Return type:: Path

validate(*, skip_schemas=False)¶

Return type:: ValidationResult

__init__(name, inputs, outputs, cwl_version='v1.2', label_text=None, doc_text=None, _base_command=<factory>, _arguments=<factory>, _requirements=<factory>, _hints=<factory>, _stdin=None, _stdout=None, _stderr=None, _intent=<factory>, _namespaces=<factory>, _schemas=<factory>, _success_codes=<factory>, _temporary_fail_codes=<factory>, _permanent_fail_codes=<factory>, _extra=<factory>)¶

class sophios.apis.python.tool_builder.CommandOutputBinding(glob=None, load_contents=None, output_eval=None, extra=<factory>)¶

A CWL output binding.

glob: Any¶

load_contents: Optional[bool]¶

output_eval: Optional[str]¶

extra: dict[str, Any]¶

to_dict()¶

Return type:: dict[str, Any]

__init__(glob=None, load_contents=None, output_eval=None, extra=<factory>)¶

class sophios.apis.python.tool_builder.Dirent(entry, entryname=None, writable=None, extra=<factory>)¶

A CWL InitialWorkDirRequirement listing entry.

entry: Any¶

entryname: Optional[str]¶

writable: Optional[bool]¶

extra: dict[str, Any]¶

to_dict()¶

Return type:: dict[str, Any]

classmethod from_input(reference, *, writable=False, entryname=None, extra=None)¶

Return type:: Dirent

__init__(entry, entryname=None, writable=None, extra=<factory>)¶

class sophios.apis.python.tool_builder.DockerRequirement(docker_pull=None, docker_load=None, docker_file=None, docker_import=None, docker_image_id=None, docker_output_directory=None, extra=<factory>)¶

DockerRequirement helper.

docker_pull: Optional[str]¶

docker_load: Optional[str]¶

docker_file: UnionType[str, dict[str, Any], None]¶

docker_import: Optional[str]¶

docker_image_id: Optional[str]¶

docker_output_directory: Optional[str]¶

extra: dict[str, Any]¶

class_name: ClassVar[str] = 'DockerRequirement'¶

to_fields()¶

Return type:: dict[str, Any]

__init__(docker_pull=None, docker_load=None, docker_file=None, docker_import=None, docker_image_id=None, docker_output_directory=None, extra=<factory>)¶

class sophios.apis.python.tool_builder.EnvironmentDef(env_name, env_value)¶

An EnvVarRequirement entry.

env_name: str¶

env_value: str¶

to_dict()¶

Return type:: dict[str, str]

__init__(env_name, env_value)¶

class sophios.apis.python.tool_builder.EnvVarRequirement(env_def, extra=<factory>)¶

EnvVarRequirement helper.

env_def: list[EnvironmentDef | dict[str, Any]]¶

extra: dict[str, Any]¶

class_name: ClassVar[str] = 'EnvVarRequirement'¶

to_fields()¶

Return type:: dict[str, Any]

__init__(env_def, extra=<factory>)¶

sophios.apis.python.tool_builder.Field¶: alias of FieldSpec

class sophios.apis.python.tool_builder.FieldSpec(type_, *, name=None, label=None, doc=None, default=<object object>, extra=None)¶

A record field definition.

type_: Any¶

name: Optional[str]¶

label_text: Optional[str]¶

doc_text: UnionType[str, list[str], None]¶

default_value: Any¶

extra: dict[str, Any]¶

__init__(type_, *, name=None, label=None, doc=None, default=<object object>, extra=None)¶

classmethod array(items, **kwargs)¶

Return type:: FieldSpec

classmethod enum(*symbols, name=None, **kwargs)¶

Return type:: FieldSpec

classmethod record(fields, *, name=None, **kwargs)¶

Return type:: FieldSpec

named(name)¶

Return type:: FieldSpec

label(text)¶

Return type:: FieldSpec

doc(text)¶

Return type:: FieldSpec

default(value)¶

Return type:: FieldSpec

to_dict()¶

Return type:: dict[str, Any]

class sophios.apis.python.tool_builder.InitialWorkDirRequirement(listing, extra=<factory>)¶

InitialWorkDirRequirement helper.

listing: Any¶

extra: dict[str, Any]¶

class_name: ClassVar[str] = 'InitialWorkDirRequirement'¶

to_fields()¶

Return type:: dict[str, Any]

__init__(listing, extra=<factory>)¶

class sophios.apis.python.tool_builder.InlineJavascriptRequirement(expression_lib=None, extra=<factory>)¶

InlineJavascriptRequirement helper.

expression_lib: Optional[list[str]]¶

extra: dict[str, Any]¶

class_name: ClassVar[str] = 'InlineJavascriptRequirement'¶

to_fields()¶

Return type:: dict[str, Any]

__init__(expression_lib=None, extra=<factory>)¶

class sophios.apis.python.tool_builder.InplaceUpdateRequirement(inplace_update=True, extra=<factory>)¶

InplaceUpdateRequirement helper.

inplace_update: bool¶

extra: dict[str, Any]¶

class_name: ClassVar[str] = 'InplaceUpdateRequirement'¶

to_fields()¶

Return type:: dict[str, Any]

__init__(inplace_update=True, extra=<factory>)¶

sophios.apis.python.tool_builder.Input¶: alias of InputSpec

class sophios.apis.python.tool_builder.InputSpec(type_, *, position=None, flag=None, required=True, separate=None, item_separator=None, value_from=None, shell_quote=None, label=None, doc=None, format=None, secondary_files=None, streamable=None, load_contents=None, load_listing=None, default=<object object>, binding_extra=None, extra=None, name=None)¶

A CWL CommandLineTool input.

type_: Any¶

position: UnionType[int, float, None]¶

flag: Optional[str]¶

required: bool¶

separate: Optional[bool]¶

item_separator: Optional[str]¶

binding_value_from: Any¶

shell_quote: Optional[bool]¶

label_text: Optional[str]¶

doc_text: UnionType[str, list[str], None]¶

format_value: Any¶

secondary_files_value: Any¶

streamable_value: Optional[bool]¶

load_contents_value: Optional[bool]¶

load_listing_value: Optional[str]¶

default_value: Any¶

binding_extra: dict[str, Any]¶

extra: dict[str, Any]¶

name: Optional[str]¶

__init__(type_, *, position=None, flag=None, required=True, separate=None, item_separator=None, value_from=None, shell_quote=None, label=None, doc=None, format=None, secondary_files=None, streamable=None, load_contents=None, load_listing=None, default=<object object>, binding_extra=None, extra=None, name=None)¶

classmethod array(items, **kwargs)¶

Return type:: InputSpec

classmethod enum(*symbols, name=None, **kwargs)¶

Return type:: InputSpec

classmethod record(fields, *, name=None, **kwargs)¶

Return type:: InputSpec

named(name)¶

Return type:: InputSpec

label(text)¶

Return type:: InputSpec

doc(text)¶

Return type:: InputSpec

default(value)¶

Return type:: InputSpec

format(value)¶

Return type:: InputSpec

secondary_files(*values)¶

Return type:: InputSpec

streamable(value)¶

Return type:: InputSpec

load_contents(value)¶

Return type:: InputSpec

load_listing(value)¶

Return type:: InputSpec

value_from(expression)¶

Return type:: InputSpec

to_dict()¶

Return type:: dict[str, Any]

class sophios.apis.python.tool_builder.Inputs(**specs)¶: Named CLT inputs. Names come from Python keyword arguments.

class sophios.apis.python.tool_builder.LoadListingRequirement(load_listing, extra=<factory>)¶

LoadListingRequirement helper.

load_listing: str¶

extra: dict[str, Any]¶

class_name: ClassVar[str] = 'LoadListingRequirement'¶

to_fields()¶

Return type:: dict[str, Any]

__init__(load_listing, extra=<factory>)¶

class sophios.apis.python.tool_builder.NetworkAccess(network_access, extra=<factory>)¶

NetworkAccess helper.

network_access: bool | str¶

extra: dict[str, Any]¶

class_name: ClassVar[str] = 'NetworkAccess'¶

to_fields()¶

Return type:: dict[str, Any]

__init__(network_access, extra=<factory>)¶

sophios.apis.python.tool_builder.Output¶: alias of OutputSpec

class sophios.apis.python.tool_builder.OutputSpec(type_, *, glob=None, from_input=None, required=True, load_contents=None, output_eval=None, label=None, doc=None, format=None, secondary_files=None, streamable=None, load_listing=None, binding_extra=None, extra=None, name=None)¶

A CWL CommandLineTool output.

type_: Any¶

required: bool¶

glob: Any¶

load_contents_value: Optional[bool]¶

output_eval: Optional[str]¶

label_text: Optional[str]¶

doc_text: UnionType[str, list[str], None]¶

format_value: Any¶

secondary_files_value: Any¶

streamable_value: Optional[bool]¶

load_listing_value: Optional[str]¶

binding_extra: dict[str, Any]¶

extra: dict[str, Any]¶

name: Optional[str]¶

__init__(type_, *, glob=None, from_input=None, required=True, load_contents=None, output_eval=None, label=None, doc=None, format=None, secondary_files=None, streamable=None, load_listing=None, binding_extra=None, extra=None, name=None)¶

classmethod array(items, **kwargs)¶

Return type:: OutputSpec

classmethod enum(*symbols, name=None, **kwargs)¶

Return type:: OutputSpec

classmethod record(fields, *, name=None, **kwargs)¶

Return type:: OutputSpec

classmethod stdout(**kwargs)¶

Return type:: OutputSpec

classmethod stderr(**kwargs)¶

Return type:: OutputSpec

named(name)¶

Return type:: OutputSpec

label(text)¶

Return type:: OutputSpec

doc(text)¶

Return type:: OutputSpec

format(value)¶

Return type:: OutputSpec

secondary_files(*values)¶

Return type:: OutputSpec

streamable(value)¶

Return type:: OutputSpec

load_listing(value)¶

Return type:: OutputSpec

load_contents(value)¶

Return type:: OutputSpec

to_dict()¶

Return type:: dict[str, Any]

class sophios.apis.python.tool_builder.Outputs(**specs)¶: Named CLT outputs. Names come from Python keyword arguments.

class sophios.apis.python.tool_builder.ResourceRequirement(cores_min=None, cores_max=None, ram_min=None, ram_max=None, tmpdir_min=None, tmpdir_max=None, outdir_min=None, outdir_max=None, extra=<factory>)¶

ResourceRequirement helper.

cores_min: UnionType[int, float, str, None]¶

cores_max: UnionType[int, float, str, None]¶

ram_min: UnionType[int, float, str, None]¶

ram_max: UnionType[int, float, str, None]¶

tmpdir_min: UnionType[int, float, str, None]¶

tmpdir_max: UnionType[int, float, str, None]¶

outdir_min: UnionType[int, float, str, None]¶

outdir_max: UnionType[int, float, str, None]¶

extra: dict[str, Any]¶

class_name: ClassVar[str] = 'ResourceRequirement'¶

to_fields()¶

Return type:: dict[str, Any]

__init__(cores_min=None, cores_max=None, ram_min=None, ram_max=None, tmpdir_min=None, tmpdir_max=None, outdir_min=None, outdir_max=None, extra=<factory>)¶

class sophios.apis.python.tool_builder.SchemaDefRequirement(types, extra=<factory>)¶

SchemaDefRequirement helper.

types: list[Any]¶

extra: dict[str, Any]¶

class_name: ClassVar[str] = 'SchemaDefRequirement'¶

to_fields()¶

Return type:: dict[str, Any]

__init__(types, extra=<factory>)¶

class sophios.apis.python.tool_builder.SecondaryFile(pattern, required=None, extra=<factory>)¶

A CWL secondary file pattern.

pattern: Any¶

required: UnionType[bool, str, None]¶

extra: dict[str, Any]¶

to_dict()¶

Return type:: str | dict[str, Any]

__init__(pattern, required=None, extra=<factory>)¶

class sophios.apis.python.tool_builder.ShellCommandRequirement(extra=<factory>)¶

ShellCommandRequirement helper.

extra: dict[str, Any]¶

class_name: ClassVar[str] = 'ShellCommandRequirement'¶

to_fields()¶

Return type:: dict[str, Any]

__init__(extra=<factory>)¶

class sophios.apis.python.tool_builder.SoftwarePackage(package, version=None, specs=None, extra=<factory>)¶

A SoftwareRequirement package entry.

package: str¶

version: Optional[list[str]]¶

specs: Optional[list[str]]¶

extra: dict[str, Any]¶

to_dict()¶

Return type:: dict[str, Any]

__init__(package, version=None, specs=None, extra=<factory>)¶

class sophios.apis.python.tool_builder.SoftwareRequirement(packages, extra=<factory>)¶

SoftwareRequirement helper.

packages: list[SoftwarePackage | dict[str, Any]]¶

extra: dict[str, Any]¶

class_name: ClassVar[str] = 'SoftwareRequirement'¶

to_fields()¶

Return type:: dict[str, Any]

__init__(packages, extra=<factory>)¶

class sophios.apis.python.tool_builder.ToolTimeLimit(timelimit, extra=<factory>)¶

ToolTimeLimit helper.

timelimit: int | str¶

extra: dict[str, Any]¶

class_name: ClassVar[str] = 'ToolTimeLimit'¶

to_fields()¶

Return type:: dict[str, Any]

__init__(timelimit, extra=<factory>)¶

class sophios.apis.python.tool_builder.ValidationResult(path, uri, process)¶

Result of validating a generated CLT with cwltool/schema-salad.

path: Path¶

uri: str¶

process: Any¶

__init__(path, uri, process)¶

class sophios.apis.python.tool_builder.WorkReuse(enable_reuse, extra=<factory>)¶

WorkReuse helper.

enable_reuse: bool | str¶

extra: dict[str, Any]¶

class_name: ClassVar[str] = 'WorkReuse'¶

to_fields()¶

Return type:: dict[str, Any]

__init__(enable_reuse, extra=<factory>)¶

sophios.apis.python.tool_builder.array_type(items)¶

Return a CWL array type expression.

Return type:: dict[str, Any]

sophios.apis.python.tool_builder.enum_type(*symbols, name=None)¶

Return a CWL enum type expression.

Return type:: dict[str, Any]

sophios.apis.python.tool_builder.record_field(type_, **kwargs)¶

Return a named CWL record field helper.

Return type:: FieldSpec

sophios.apis.python.tool_builder.record_type(fields, *, name=None)¶

Return a CWL record type expression.

Return type:: dict[str, Any]

sophios.apis.python.tool_builder.secondary_file(pattern, *, required=None, **extra)¶

Create a secondary file specification.

Return type:: SecondaryFile

sophios.apis.python.tool_builder.step_from_command_line_tool(tool, *, step_name=None, run_path=None, config=None, tool_registry=None)¶

Convert a built CLT into a workflow Step entirely in memory.

Parameters:

tool (CommandLineTool) – Built CLT to wrap as a workflow step.
step_name (str | None) – Optional workflow step name override.
run_path (str | Path | None) – Optional virtual .cwl path for compiler bookkeeping.
config (dict[str, Any] | None) – Optional input values to pre-bind.
tool_registry (Tools | None) – Optional tool registry retained on the step.

Returns:

A workflow step backed by the CLT without touching disk.

Return type:

Step

sophios.apis.python.tool_builder.validate_cwl_document(document, *, filename='tool.cwl', skip_schemas=False)¶

Validate a generated CLT document through cwltool/schema-salad.

Return type:: ValidationResult

sophios.compute_payload¶

Schema-backed compute-slurm payload objects.

class sophios.compute_payload.ComputeConfig(toil=None, output=None, slurm=None)¶

Schema mirror for computeConfig.

toil: Optional[ToilConfig]¶

output: Optional[OutputConfig]¶

slurm: Optional[SlurmConfig]¶

to_dict()¶

Render nested compute configuration.

Returns:: JSON-ready computeConfig.
Return type:: Json

__init__(toil=None, output=None, slurm=None)¶

exception sophios.compute_payload.ComputePayloadValidationError¶: Raised when a compute payload does not match the checked-in schema.

class sophios.compute_payload.ComputeWorkflowPayload(cwl_workflow, cwl_job_inputs, workflow_id=None, jobs=<factory>, compute_config=None)¶

Schema-backed compute-slurm request payload.

cwl_workflow: Dict[str, Any]¶

cwl_job_inputs: Dict[str, Any]¶

workflow_id: Optional[str]¶

jobs: Dict[str, Any]¶

compute_config: Optional[ComputeConfig]¶

get_compute_payload()¶

Render and validate the compute request payload.

Raises:: ComputePayloadValidationError – If the rendered payload is invalid.
Returns:: Schema-valid compute payload.
Return type:: Json

__init__(cwl_workflow, cwl_job_inputs, workflow_id=None, jobs=<factory>, compute_config=None)¶

class sophios.compute_payload.OutputConfig(mode=None, output_dir=None)¶

Schema mirror for computeConfig.outputConfig.

mode: Optional[str]¶

output_dir: UnionType[str, Path, None]¶

classmethod service_default()¶

Use the service-managed output directory.

Returns:: Service-default output configuration.
Return type:: OutputConfig

classmethod workflow_declared()¶

Preserve the workflow’s own output behavior.

Returns:: Workflow-declared output configuration.
Return type:: OutputConfig

classmethod user_specified(output_dir)¶

Use a caller-provided output directory.

Parameters:: output_dir (str | Path) – Directory that compute-slurm should use.
Returns:: User-specified output configuration.
Return type:: OutputConfig

classmethod from_json(*, mode=None, outputDir=None)¶

Construct from schema-shaped JSON field names.

Parameters:

mode (str | None) – Raw schema mode value such as workflowDeclared.
outputDir (str | Path | None) – Raw schema outputDir value.

Returns:

Output configuration using JSON/schema naming.

Return type:

OutputConfig

to_dict()¶

Render the output configuration.

Raises:: ValueError – If mode=’userSpecified’ is missing output_dir.
Returns:: JSON-ready output configuration.
Return type:: Json

__init__(mode=None, output_dir=None)¶

class sophios.compute_payload.SlurmConfig(job_name=None, partition=None, slurm_job_gpu_count=None, cpus_per_task=None, nodes=None, tasks_per_node=None, output=None, error=None, time_limit=None, memory=None)¶

Schema mirror for computeConfig.slurmConfig.

job_name: Optional[str]¶

partition: Optional[str]¶

slurm_job_gpu_count: Optional[int]¶

cpus_per_task: Optional[int]¶

nodes: Optional[int]¶

tasks_per_node: Optional[int]¶

output: Optional[str]¶

error: Optional[str]¶

time_limit: Optional[str]¶

memory: Optional[str]¶

to_dict()¶

Render the SLURM configuration.

Returns:: JSON-ready SLURM configuration.
Return type:: Json

__init__(job_name=None, partition=None, slurm_job_gpu_count=None, cpus_per_task=None, nodes=None, tasks_per_node=None, output=None, error=None, time_limit=None, memory=None)¶

class sophios.compute_payload.ToilConfig(log_level=None)¶

Schema mirror for computeConfig.toilConfig.

log_level: Optional[str]¶

to_dict()¶

Render the toil configuration.

Returns:: JSON-ready toil configuration.
Return type:: Json

__init__(log_level=None)¶

sophios.compute_payload.validate_compute_payload(payload)¶

Validate a compute payload mapping against the checked-in schema.

Parameters:: payload (Mapping[str, Any]) – Candidate compute payload.
Raises:: ComputePayloadValidationError – If the payload is invalid.
Returns:: Schema-valid compute payload.
Return type:: Json

sophios.compute_submit¶

sophios.compute_submit.submit_compute_json(payload_json, submit_url, *, timeout=(5, 30), poll_interval_seconds=15, log_path=None)¶

Submit an already-rendered compute payload JSON object.

Return type:: int

sophios.compute_submit.submit_compute_payload(payload, submit_url, *, timeout=(5, 30), poll_interval_seconds=15, log_path=None)¶

Submit a compute payload and wait for the job to start.

Return type:: int

Python API Reference¶

sophios.apis.python.workflow and sophios.apis.python.tool_builder¶

sophios.apis.python.workflow¶

sophios.apis.python.tool_builder¶

sophios.compute_payload¶

sophios.compute_submit¶

Sophios

Navigation

Related Topics