Workflow

junjo.Workflow

Kind: Class

SDK version: 0.69.0

Documentation channel: Stable release

Signature

Workflow(graph_factory: GraphFactory[Graph], store_factory: StoreFactory[StoreT], max_iterations: int = 100, hooks: Hooks | None = None, name: str | None = None)

View source

Constructor

A Workflow is a top-level, executable collection of nodes and edges arranged as a graph. It manages its own state and store, distinct from any parent or sub-workflows.

This class is generic and requires two type parameters for a convenient and type-safe developer experience:

StateT is the type of state managed by this workflow and should be a subclass of BaseState. StoreT is the store type used by this workflow and should be a subclass of BaseStore.

Every call to Workflow.execute creates a fresh graph, a fresh store, and a fresh execution context. That makes a Workflow instance a reusable definition or blueprint rather than a mutable live-run container.

Parameters

Name	Type	Description	Default
name	(str | None, optional)	An optional name for the workflow. If not provided, the class name is used.	None
graph_factory	GraphFactory[Graph]	A callable that returns a new instance of the workflow’s graph (Graph). This factory is invoked at the beginning of each Workflow.execute call to ensure a fresh, isolated graph for that execution.
store_factory	StoreFactory[StoreT]	A callable that returns a new instance of the workflow’s store (StoreT). This factory is invoked at the beginning of each Workflow.execute call to ensure a fresh store for that execution.
max_iterations	(int, optional)	The maximum number of times any single node or subflow may run within one workflow execution. For RunConcurrent, Junjo counts each child item in the concurrent group rather than the RunConcurrent wrapper itself. This helps detect accidental loops. Defaults to 100.	100
hooks	(Hooks | None, optional)	An optional Hooks registry for observing workflow lifecycle events. Hooks are optional observers; they do not control OpenTelemetry instrumentation or workflow execution.	None

Example without hooks

workflow = Workflow[MyState, MyStore](
    name="demo_base_workflow",
    graph_factory=create_my_graph,
    store_factory=lambda: MyStore(initial_state=MyState()),
)

result = await workflow.execute()
print(result.state.model_dump_json())

Example with hooks

import logging

hooks = Hooks()
logger = logging.getLogger(__name__)

def log_completed(event) -> None:
    logger.info(
        "workflow completed %s %s %s",
        event.name,
        event.run_id,
        event.result.state.model_dump(),
    )

hooks.on_workflow_completed(log_completed)

workflow = Workflow[MyState, MyStore](
    name="configured_workflow",
    graph_factory=create_my_graph,
    store_factory=lambda: MyStore(initial_state=MyState()),
    hooks=hooks,
)

Passing Parameters to Factories

To provide parameters to your graph_factory or store_factory when you create a workflow, wrap the factory call in a lambda. This creates an argument-less factory that closes over the dependencies you want to inject while preserving the fresh-per-run execution model.

def create_graph_with_dependency(emulator: Emulator) -> Graph:
    return Graph(...)

my_emulator = Emulator()

workflow = Workflow[MyState, MyStore](
    name="configured_workflow",
    graph_factory=lambda: create_graph_with_dependency(my_emulator),
    store_factory=lambda: MyStore(initial_state=MyState()),
)

Members

max_iterations

max_iterations = max_iterations

View source

Public attribute.

hooks

hooks = hooks

View source

Public attribute.

id

id: str

View source

Returns the stable definition identifier for this workflow object.

name

name: str

View source

Returns the configured workflow name or the class name.

span_type

span_type: ExecutableType

View source

Returns the OpenTelemetry span type for this executable.

execute

execute(parent_store: ParentStoreT | None = None, parent_id: str | None = None, *, validate_graph: bool = True, store: StoreT | None = None, correlation: ExecutionCorrelation | None = None) -> ExecutionResult[StateT]

View source

Execute the workflow or subflow and return the final execution snapshot.

Each call creates a fresh graph, a fresh run id, and a fresh lifecycle dispatcher. A Workflow uses the supplied Store or calls its factory. Subflows always create their own Store. This keeps the workflow definition itself immutable and safe to reuse across concurrent runs.

Parameters

Name	Type	Description	Default
parent_store	ParentStoreT | None	The parent store when executing a subflow. Top-level workflows should omit this argument.	None
parent_id	str | None	The parent workflow or subflow identifier when nested.	None
store	StoreT | None	Optional live application Store for a Workflow. When supplied, the Store factory is not called. Subflows do not accept this argument; use their existing pre/post mapping actions.	None
validate_graph	bool	Whether to run Graph.validate() on the fresh graph before execution starts. Defaults to True.	True
correlation	ExecutionCorrelation | None	Optional trusted application identity for this execution tree. Nested Junjo executables inherit it and cannot replace it.	None

Returns

• ExecutionResult[StateT]: A detached snapshot of the completed execution, including the final state and current-scope execution counts.

Raises

Exception	Description
WorkflowExecutionError	If an admitted execution fails. The original failure is available as error.__cause__.
WorkflowCancelledError	If an admitted execution is cancelled. This remains an CancelledError.