BaseStore

junjo.BaseStore

Kind: Class

SDK version: 0.69.0

Documentation channel: Stable release

Signature

BaseStore(initial_state: StateT)

View source

BaseStore represents a generic store for managing the state of a workflow. It is designed to be subclassed with a specific state type (a Pydantic model derived from BaseState).

The store is responsible for:

Managing the state of a workflow or subflow execution.
Making immutable updates to that state safely in a concurrent environment.
Validating committed updates against the underlying Pydantic model.

The store uses an Lock to ensure that state updates are concurrency-safe. Each committed update is derived, validated, and applied against the exact locked state version it modifies, which prevents stale validate-then-apply races under concurrent execution.

Subclass BaseStore with your own state type and expose domain-specific actions that call set_state.

Example

class MyWorkflowState(BaseState):
    user_input: str
    processed_data: dict | None = None

class MyWorkflowStore(BaseStore[MyWorkflowState]):
    async def set_processed_data(self, data: dict) -> None:
        await self.set_state({"processed_data": data})

Constructor

Parameters

Name	Type	Description	Default
`initial_state`	`StateT`	The initial state of the store.

Members

`id`

id: str

View source

Returns the unique identifier of this store instance.

`get_state`

get_state() -> StateT

View source

Return a detached deep copy of the current state.

This method follows the immutability principle for read access: callers receive a deep snapshot that can be inspected freely without mutating the live store. Store updates must still flow through store actions and set_state.

`get_state_json`

get_state_json() -> str

View source

Return the current state as a JSON string.

This is useful for logging, tracing, or serializing the current state without manually calling model_dump_json on a snapshot.

`set_state`

set_state(update: dict) -> None

View source

Update the store’s state with a dictionary of changes.

The update is shallow-merged with the current locked state at the top-level field boundary, validated against the store’s Pydantic model, and committed atomically if it changes the state. Nested mappings or nested models are replaced as field values; Junjo does not recursively deep-merge nested structures. This method also emits OpenTelemetry set_state events and lifecycle state-changed hooks when a commit occurs.

Incoming mutable values become Store-owned copies before validation. Later changes to the caller’s objects cannot change committed state, and validators do not modify those original input objects.

Parameters

Name	Type	Description	Default
`update`	`dict`	A dictionary of updates to apply to the state.

Raises

Exception	Description
`ValueError`	If model validation rejects the update; no state is committed.

class MessageWorkflowState(BaseState):
    received_message: Message

class MessageWorkflowStore(BaseStore[MessageWorkflowState]):
    async def set_received_message(self, payload: Message) -> None:
        await self.set_state({"received_message": payload})

payload = Message(...)
await store.set_received_message(payload)