matflow.InputSource#

class matflow.InputSource(source_type, import_ref=None, task_ref=None, task_source_type=None, element_iters=None, path=None, where=None)#

Bases: InputSource

An input source to a workflow task.

Parameters:

  • source_type (InputSourceType) – Type of the input source.

  • import_ref (int | None) – Where the input comes from when the type is IMPORT.

  • task_ref (int | None) – Which task is this an input for? Used when the type is TASK.

  • task_source_type (TaskSourceType) – Type of task source.

  • element_iters (list[int] | None) – Which element iterations does this apply to?

  • path (str | None) – Path to where this input goes.

  • where (Rule | list[Rule] | ElementFilter) – Filtering rules.

Methods

default

Make an instance of an input source that is default.

from_json_like

Make an instance of this class from JSON (or YAML) data.

from_string

Parse a dot-delimited string definition of an InputSource.

get_task

If source_type is task, then return the referenced task from the given workflow.

import_

Make an instance of an input source that is an import.

is_in

Check if this input source is in a list of other input sources, without considering the element_iters and where attributes.

local

Make an instance of an input source that is local.

task

Make an instance of an input source that is a task.

to_dict

Serialize this object as a dictionary.

to_json_like

Serialize this object as an object structure that can be trivially converted to JSON.

to_string

Render this input source as a string.

Attributes

where

Filtering rules.

source_type

Type of the input source.

import_ref

Where the input comes from when the type is IMPORT.

task_ref

Which task is this an input for? Used when the type is TASK.

task_source_type

Type of task source.

element_iters

Which element iterations does this apply to?

path

Path to where this input goes.
classmethod default()#

Make an instance of an input source that is default.

Return type:

Self

element_iters#

Which element iterations does this apply to?

classmethod from_json_like(json_like, shared_data=None)#

Make an instance of this class from JSON (or YAML) data.

Parameters:

  • json_like – The data to deserialise.

  • shared_data – Shared context data.

Return type:

The deserialised object.

classmethod from_string(str_defn)#

Parse a dot-delimited string definition of an InputSource.

Parameter#

str_defn:

The string to parse.

Examples

For a local task input source, use:

>>> InputSource.from_string("local")

For a schema input default source, use:

>>> InputSource.from_string("default")

For task input sources, specify either the task insert ID (typically this is just the task index within the workflow), or the task’s unique name, which is usually just the associated task schema’s objective, but if multiple tasks use the same schema, it will be suffixed by an index, starting from one.

>>> InputSource.from_string("task.0.input")
>>> InputSource.from_string("task.my_task.input")
Parameters:

str_defn (str) –

Return type:

Self

get_task(workflow)#

If source_type is task, then return the referenced task from the given workflow.

Parameters:

workflow (Workflow) –

Return type:

WorkflowTask | None

classmethod import_(import_ref, element_iters=None, where=None)#

Make an instance of an input source that is an import.

Parameters:

  • import_ref (int) – Import reference.

  • element_iters (list[int] | None) – Originating element iterations.

  • where (Where | None) – Filtering rule.

Return type:

Self

import_ref#

Where the input comes from when the type is IMPORT.

is_in(other_input_sources)#

Check if this input source is in a list of other input sources, without considering the element_iters and where attributes.

Parameters:

other_input_sources (Sequence[InputSource]) –

Return type:

int | None

classmethod local()#

Make an instance of an input source that is local.

Return type:

Self

path#

Path to where this input goes.

source_type#

Type of the input source.

classmethod task(task_ref, task_source_type=None, element_iters=None, where=None)#

Make an instance of an input source that is a task.

Parameters:

  • task_ref (int) – Source task reference.

  • task_source_type (TaskSourceType | str | None) – Type of task source.

  • element_iters (list[int] | None) – Originating element iterations.

  • where (Where | None) – Filtering rule.

Return type:

Self

task_ref#

Which task is this an input for? Used when the type is TASK.

task_source_type#

Type of task source.

to_dict()#

Serialize this object as a dictionary.

Return type:

dict[str, Any]

to_json_like(dct=None, shared_data=None, exclude=(), path=None)#

Serialize this object as an object structure that can be trivially converted to JSON. Note that YAML can also be produced from the result of this method; it just requires a different final serialization step.

Parameters:

  • dct (dict[str, JSONable] | None) –

  • shared_data (_JSONDeserState) –

  • exclude (Container[str | None]) –

  • path (list | None) –

Return type:

tuple[JSONDocument, _JSONDeserState]

to_string()#

Render this input source as a string.

Return type:

str

where: ElementFilter | None#

Filtering rules.