primus.cli.procedure.tactus_adapters.storage module

Primus Storage Adapter for Tactus.

Implements the Tactus StorageBackend protocol using Primus GraphQL API. The Procedure record is an index card: it holds legacy-compatible object keys and replay_index. All procedure data (state, lua_state, checkpoints) lives in application-authorized GraphQL artifact attachments.

class primus.cli.procedure.tactus_adapters.storage.PrimusStorageAdapter(client, procedure_id: str, *, artifact_store=None)

Bases: object

Implements Tactus StorageBackend protocol using Primus GraphQL.

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

Initialize Primus storage adapter.

Args:

client: PrimusDashboardClient instance procedure_id: ID of the procedure

__init__(client, procedure_id: str, *, artifact_store=None)

Initialize Primus storage adapter.

Args:

client: PrimusDashboardClient 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 verified procedure state and checkpoints through artifact tickets.

save_procedure_metadata(procedure_id: str, metadata: ProcedureMetadata) → None

Save procedure metadata to Primus via GraphQL.

Args:

procedure_id: Procedure ID (for API compatibility) 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 Primus.

Args:

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

exception primus.cli.procedure.tactus_adapters.storage.ProcedureArtifactStorageError

Bases: RuntimeError

Procedure persistence could not load or store a verified artifact.

primus.cli.procedure.tactus_adapters.storage.download_procedure_attachment(client, procedure_id: str, filename: str, pointer: Any, *, content_type: str = 'text/plain', artifact_store=None) → bytes

Load an integrity-verified procedure attachment through GraphQL tickets.

primus.cli.procedure.tactus_adapters.storage.upload_procedure_attachment(client, procedure_id: str, filename: str, content: str | bytes, *, content_type: str, existing_metadata: Dict[str, Any] | None = None, artifact_store=None) → Dict[str, Any]

Store procedure code or an attachment through an authorized ticket.