Subflow

junjo.Subflow

Kind: Class

SDK version: 0.69.0

Documentation channel: Stable release

Signature

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

View source

Constructor

A Subflow is a workflow that:

• Executes within a parent workflow or parent subflow.
• Has its own isolated state and store.
• Can interact with its parent workflow state before and after execution via pre_run_actions and post_run_actions.

Like top-level workflows, subflows create a fresh graph and a fresh store for every execution. The child run is isolated from the parent store except for the explicit handoff points provided by the pre- and post-run hooks.

Parameters

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

Example

class ExampleSubflow(
    Subflow[SubflowState, SubflowStore, ParentState, ParentStore]
):
    async def pre_run_actions(self, parent_store, subflow_store):
        parent_state = await parent_store.get_state()
        await subflow_store.set_parameter(
            {"parameter": parent_state.parameter}
        )

    async def post_run_actions(self, parent_store, subflow_store):
        subflow_state = await subflow_store.get_state()
        await parent_store.set_subflow_result(subflow_state.result)

example_subflow = ExampleSubflow(
    graph_factory=create_example_subflow_graph,
    store_factory=lambda: ExampleSubflowStore(
        initial_state=ExampleSubflowState()
    ),
)

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.

pre_run_actions

pre_run_actions(parent_store: ParentStoreT, subflow_store: StoreT) -> None

View source

This method is called before the subflow has run.

This is where you can pass initial state values from the parent workflow to the subflow store for this specific run.

Parameters

Name	Type	Description	Default
parent_store	ParentStoreT	The parent store to interact with.
subflow_store	StoreT	The store for this specific subflow execution.

post_run_actions

post_run_actions(parent_store: ParentStoreT, subflow_store: StoreT) -> None

View source

This method is called after the subflow has run.

This is where you can update the parent store with the results of the child workflow.

Parameters

Name	Type	Description	Default
parent_store	ParentStoreT	The parent store to update.
subflow_store	StoreT	The store for this specific subflow execution.