`tierkreis.builder`¶

Typed graph builder for Tierkreis workflows.

Module Contents¶

Classes¶

`TList`	A list of models.
`Function`	A worker function type.
`TypedGraphRef`	A typed tierkreis graph.
`LoopOutput`	Protocol for loop output models to ensure should continue.
`Graph`	Class to construct typed workflow graphs.

Functions¶

script

Add a script to the graph.

API¶

class TList¶: A list of models.

class Function¶

Bases: tierkreis.controller.data.models.TNamedModel, typing.Protocol

A worker function type.

Abstract:

property namespace: str¶

The namespace name.

Returns:: The namespace name.
Return type:: str

static out() → type[Out]¶

Return the output type of the function.

Returns:: The output type.
Return type:: type[Out]

__slots__ = ()¶

classmethod __init_subclass__(*args, **kwargs)¶

classmethod __class_getitem__(item)¶

class TypedGraphRef¶

A typed tierkreis graph.

Attr graph_ref:: The graph reference.
Attr outputs_type:: The output type of the graph.

graph_ref: tierkreis.controller.data.models.TKR[tierkreis.controller.data.types.Workflow[tierkreis.builder.TypedGraphRef.Ins, tierkreis.builder.TypedGraphRef.Outs]] = None¶

outputs_type: type[tierkreis.builder.TypedGraphRef.Outs] = None¶

class LoopOutput¶

Bases: tierkreis.controller.data.models.TNamedModel, typing.Protocol

Protocol for loop output models to ensure should continue.

property should_continue: tierkreis.controller.data.models.TKR[bool]¶

The loop continuation port.

Returns:: The continuation port value.
Return type:: TKR[bool]

__slots__ = ()¶

classmethod __init_subclass__(*args, **kwargs)¶

classmethod __class_getitem__(item)¶

script(script_name: str, script_input: tierkreis.controller.data.models.TKR[bytes]) → tierkreis.builder.Function[tierkreis.controller.data.models.TKR[bytes]]¶

Add a script to the graph.

A shell script or binary with a single input and output. Inputs are provided from the standard input and outputs to the standard output.

Parameters:

script_name (str) – The name of the script.
script_input (TKR[bytes]) – The input to the script.

Returns:

The output of the script.

Return type:

Function[TKR[bytes]]

class Graph(inputs_type: type[Inputs] = EmptyModel, outputs_type: type[Outputs] = EmptyModel, breakpoints_on_inputs: bool = False, breakpoints_on_outputs: bool = False)¶

Class to construct typed workflow graphs.

Fields:: data (GraphData): The underlying graph data. inputs_type (type[TModel]) The input type of the graph. inputs (inputs_type): The inputs to the graph. outputs_type((type[TModel])): The output type of the graph. _breakpoints_on_outputs(bool): Make outputs act as breakpoints.

Initialization

inputs_type: type[tierkreis.builder.Graph.Inputs] = None¶

outputs_type: type[tierkreis.builder.Graph.Outputs] = None¶

inputs: tierkreis.builder.Graph.Inputs = None¶

get_data() → tierkreis.controller.data.graph.GraphData¶

Return the underlying graph from the builder.

Returns:: The graph.
Return type:: GraphData

ref() → tierkreis.builder.TypedGraphRef[Inputs, Outputs]¶

Return a reference of the typed graph.

Returns:: The ref of the typed graph.
Return type:: TypedGraphRef[Inputs, Outputs]

finish_with_outputs(outputs: Outputs) → tierkreis.controller.data.types.Workflow[Inputs, Outputs]¶

Set output nodes of a graph.

Parameters:: outputs (Outputs) – The output nodes.

embed(other_fg: tierkreis.controller.data.types.Workflow[tierkreis.builder.Graph.embed.A, tierkreis.builder.Graph.embed.B], inputs: tierkreis.builder.Graph.embed.A, outputs_type: type[tierkreis.builder.Graph.embed.B]) → tierkreis.builder.Graph.embed.B¶

const(value: tierkreis.builder.Graph.const.T) → tierkreis.controller.data.models.TKR[tierkreis.builder.Graph.const.T]¶

Add a constant node to the graph.

Note if the value is a Workflow, use graph_const instead to preserve type information.

Returns:: The constant value.
Return type:: TKR[T]

ifelse(pred: tierkreis.controller.data.models.TKR[bool], if_true: tierkreis.controller.data.models.TKR[tierkreis.builder.Graph.ifelse.A], if_false: tierkreis.controller.data.models.TKR[tierkreis.builder.Graph.ifelse.B], **kwargs) → tierkreis.controller.data.models.TKR[tierkreis.builder.Graph.ifelse.A] | tierkreis.controller.data.models.TKR[tierkreis.builder.Graph.ifelse.B]¶

Add an if-else node to the graph.

This will be evaluated lazily. The values can be returned from an eval node or another graph.

Parameters:

pred (TKR[bool]) – The predicate value.
if_true (TKR[A]) – The value if the predicate is true.
if_false (TKR[B]) – The value if the predicate is false.
kwargs (dict) – Optional metadata for the node (e.g., has_breakpoint=True)

Returns:

The outputs of the if-else expression.

Return type:

TKR[A] | TKR[B]

eifelse(pred: tierkreis.controller.data.models.TKR[bool], if_true: tierkreis.controller.data.models.TKR[tierkreis.builder.Graph.eifelse.A], if_false: tierkreis.controller.data.models.TKR[tierkreis.builder.Graph.eifelse.B], **kwargs) → tierkreis.controller.data.models.TKR[tierkreis.builder.Graph.eifelse.A] | tierkreis.controller.data.models.TKR[tierkreis.builder.Graph.eifelse.B]¶

Add an eager if-else node to the graph.

This will be evaluated eagerly. The values can be returned from an eval node or another graph.

Parameters:

pred (TKR[bool]) – The predicate value.
if_true (TKR[A]) – The value if the predicate is true.
if_false (TKR[B]) – The value if the predicate is false.
kwargs (dict) – Optional metadata for the node (e.g., has_breakpoint=True)

Returns:

The outputs of the if-else expression.

Return type:

TKR[A] | TKR[B]

graph_const(graph: tierkreis.controller.data.types.Workflow[tierkreis.builder.Graph.graph_const.A, tierkreis.builder.Graph.graph_const.B]) → tierkreis.builder.TypedGraphRef[tierkreis.builder.Graph.graph_const.A, tierkreis.builder.Graph.graph_const.B]¶

Add a graph constant to the graph. Returns a typed reference to the graph that can be used in eval and loop nodes.

Parameters:: graph – The graph to add as a constant.

task(func: tierkreis.builder.Function[tierkreis.builder.Graph.task.Out], **kwargs) → tierkreis.builder.Graph.task.Out¶

Add a worker task node to the graph.

Parameters:

func (Function[Out]) – The worker function.
kwargs (dict) – Optional metadata for the node (e.g., has_breakpoint=True)

Returns:

The outputs of the task.

Return type:

Out

eval(body: tierkreis.controller.data.types.Workflow[tierkreis.builder.Graph.eval.A, tierkreis.builder.Graph.eval.B] | tierkreis.builder.TypedGraphRef[tierkreis.builder.Graph.eval.A, tierkreis.builder.Graph.eval.B], eval_inputs: tierkreis.builder.Graph.eval.A, **kwargs) → tierkreis.builder.Graph.eval.B¶

Add a evaluation node to the graph.

This will evaluate a nested graph with the given inputs.

Parameters:

body (TypedGraphRef[A, B] | Workflow[A, B], where A are the input type and B the output type of the graph.) – The graph to evaluate.
eval_inputs (A) – The inputs to the graph.
kwargs (dict) – Optional metadata for the node (e.g., has_breakpoint=True)

Returns:

The outputs of the evaluation.

Return type:

loop(body: tierkreis.builder.TypedGraphRef[tierkreis.builder.Graph.loop.A, tierkreis.builder.Graph.loop.B] | tierkreis.controller.data.types.Workflow[tierkreis.builder.Graph.loop.A, tierkreis.builder.Graph.loop.B], loop_inputs: tierkreis.builder.Graph.loop.A, name: str | None = None, **kwargs) → tierkreis.builder.Graph.loop.B¶

Add a loop node to the graph.

This will loop over the given graph until the should_continue output is false. To trace intermediate values, use the name attribute in conjunction with read_loop_trace.

Parameters:

body (TypedGraphRef[A, B] | Workflow[A, B], where A are the input type and B the output type of the graph.) – The graph to loop.
loop_inputs (A) – The inputs to the loop graph.
name (str | None) – An optional name for the loop.
kwargs (dict) – Optional metadata for the node (e.g., has_breakpoint=True)

Returns:

The outputs of the loop.

Return type:

map(body: tierkreis.builder.TypedGraphRef | collections.abc.Callable | tierkreis.controller.data.types.Workflow, map_inputs: tierkreis.controller.data.models.TKR | tierkreis.builder.TList, **kwargs) → Any¶

Add a map node to the graph.

Parameters:

body (TypedGraphRef | Callable | Workflow) – The graph to map over.
map_inputs (TKR | TList) – The values to map over.
kwargs (dict) – Optional metadata for the node (e.g., has_breakpoint=True)

Returns:

The outputs of the map.

Return type:

Any

tierkreis.builder¶

Module Contents¶

Classes¶

Functions¶

API¶

`tierkreis.builder`¶