plexus.cli.procedure.tactus_adapters.storage module

Plexus Storage Adapter for Tactus.

Implements the Tactus StorageBackend protocol using Plexus GraphQL API. Stores checkpoints and state in the Procedure.metadata JSON field.

class plexus.cli.procedure.tactus_adapters.storage.PlexusStorageAdapter(client, procedure_id: str)

Bases: object

Implements Tactus StorageBackend protocol using Plexus GraphQL.

Stores all procedure data (checkpoints, state, lua_state) in the Procedure.metadata JSON field via GraphQL mutations.

Initialize Plexus storage adapter.

Args:

client: PlexusDashboardClient instance procedure_id: ID of the procedure

__init__(client, procedure_id: str)

Initialize Plexus storage adapter.

Args:

client: PlexusDashboardClient instance procedure_id: ID of the procedure

checkpoint_clear_after(procedure_id: str, name: str) → None

Clear checkpoint and all subsequent ones.

checkpoint_clear_all(procedure_id: str) → None

Clear all checkpoints (but preserve state).

checkpoint_exists(procedure_id: str, name: str) → bool

Check if checkpoint exists.

checkpoint_get(procedure_id: str, name: str) → Any | None

Get checkpoint value.

checkpoint_save(procedure_id: str, name: str, result: Any) → None

Save a checkpoint.

get_state(procedure_id: str) → Dict[str, Any]

Get mutable state dictionary.

load_procedure_metadata(procedure_id: str) → ProcedureMetadata

Load procedure metadata from Plexus via GraphQL.

Args:

procedure_id: Procedure ID to load

Returns:

ProcedureMetadata with checkpoints, state, and lua_state

save_procedure_metadata(metadata: ProcedureMetadata) → None

Save procedure metadata to Plexus via GraphQL.

Args:

metadata: ProcedureMetadata to save

set_state(procedure_id: str, state: Dict[str, Any]) → None

Set mutable state dictionary.

state_clear(procedure_id: str) → None

Clear all state.

state_delete(procedure_id: str, key: str) → None

Delete state key.

state_get(procedure_id: str, key: str, default: Any = None) → Any

Get state value.

state_set(procedure_id: str, key: str, value: Any) → None

Set state value.

update_procedure_status(procedure_id: str, status: str, waiting_on_message_id: str | None = None) → None

Update procedure status in Plexus.

Args:

procedure_id: Procedure ID status: New status waiting_on_message_id: Optional message ID if waiting for human