Workflows

Workflow

At the basic level a workflow is an object that holds a series of nodes to be called in sequence. The workflow object also includes helper methods to generate and append the nodes defined in the Builtin Nodes section of the documentation.

Just like every node in pyApp Flow a workflow is called with an pyapp_flow.WorkflowContext object, this means workflows can be nested in workflows, or called from a pyapp_flow.ForEach node.

The one key aspect with a workflow object is related to context variable scope. When a workflow is triggered the context scope is copied and any changes made to the variables are discarded when the workflow ends. However, just like Python scoping only the reference to the variable is copied meaning mutable objects can be modified (eg list/dicts).

workflow = (
    Workflow(name="My Workflow")
    .nodes(...)
)

class pyapp_flow.Workflow(name: str, description: str | None = None)

A collection of Nodes that make up a workflow.

Parameters:

• name – The name of the workflow

• description – An optional description (similar to doc text) for the workflow.

execute(context: WorkflowContext | None = None, *, dry_run: bool = False, **context_vars) → WorkflowContext

Execute workflow.

This is the main way to trigger a work flow.

Parameters:

• context – Optional context; a new one will be created if not supplied.

• dry_run – Flag used to skip steps that have side effects.

• context_vars – Key/Value pairs to initialise the context with.

Returns:

The context used to execute the workflow

property name

Name of object

nested(*nodes_: Callable[[WorkflowContext], Any]) → Workflow

Add nested node(s), nested nodes have their own scope.

Parameters:

nodes – Collection of nodes call from nested block.

Returns:

Returns self; fluent interface

nodes(*nodes_: Callable[[WorkflowContext], Any]) → Workflow

Append additional node(s) into the node list.

Parameters:

nodes – Nodes to append to the current block

Returns:

Returns self; fluent interface

set_vars(**kwargs) → Workflow

Set variables to a particular value.

Parameters:

kwargs – Key/Value pairs to update in the context

Returns:

Returns self; fluent interface

WorkflowContext

The workflow context object holds the state of the workflow including handling variable scoping and helper methods for logging progress.

class pyapp_flow.WorkflowContext(logger: Logger | None = None, **variables)

Current context of the workflow.

This object can be used as a context object to apply state scoping.

The current state can be modified via the state property, this is a dict of Key/Value pairs.

Parameters:

• logger – And optional logger; one will be created named pyapp_flow if not provided.

• variables – Initial state of variables.

context = WorkflowContext(foo="123")

with context:
    assert context.state["foo"] == "123"
    context.state["bar"] == "456"

assert "bar" not in context.state

capture_trace(*, force: bool = False)

Capture trace and location of an exception.

Unless forced will not overwrite an existing trace.

debug(msg, *args)

Write a debug message to log.

property depth: int

Current scope depth; or size of the current state vector

error(msg, *args)

Write an error message to log.

exception(msg, *args)

Write an exception message to log.

property flow_trace: FlowTrace | None

Return the current flow trace.

format(message: str) → str

Format a message using context variables.

Return a formatted version of message, using substitutions context variables. The substitutions are identified by braces (‘{’ and ‘}’).

property indent: str

Spacing based on the current indent level.

info(msg, *args)

Write an info message to log.

log(level: int, msg: str, *args, **kwargs)

Log a message to logger indented by the current scope depth.

pop_state()

Pop the top state from the state vector.

push_state()

Push a new state onto the stack.

set_trace_args(args)

Set the branch args for the current trace node.

trace(node: Navigable)

Add a node to the trace.

warning(msg, *args)

Write a warning message to log.