Persistence

WorkflowStore interface package.

class stabilize.persistence.store.WorkflowStore[source]

Bases: ABC

Abstract interface for execution persistence.

abstractmethod store(execution)[source]

Store a complete execution.

Creates the execution and all its stages.

Parameters:

execution (Workflow) – The execution to store

Return type:

None

abstractmethod retrieve(execution_id)[source]

Retrieve an execution by ID.

Parameters:

execution_id (str) – The execution ID

Returns:

The execution

Raises:

WorkflowNotFoundError – If not found

Return type:

Workflow

abstractmethod retrieve_execution_summary(execution_id)[source]

Retrieve execution metadata without stages.

Parameters:

execution_id (str) – The execution ID

Returns:

The execution with empty stages list

Raises:

WorkflowNotFoundError – If not found

Return type:

Workflow

abstractmethod update_status(execution)[source]

Update the status of an execution.

Parameters:

execution (Workflow) – The execution with updated status

Return type:

None

abstractmethod delete(execution_id)[source]

Delete an execution and all its stages.

Parameters:

execution_id (str) – The execution ID

Return type:

None

abstractmethod store_stage(stage, expected_phase=None)[source]

Store or update a stage.

Parameters:

  • stage (StageExecution) – The stage to store

  • expected_phase (str | None) – If provided, the WHERE clause includes status = expected_phase. This enables phase-aware optimistic locking where updates only succeed if the stage is in the expected status.

Return type:

None

abstractmethod add_stage(stage)[source]

Add a new stage to an execution.

Parameters:

stage (StageExecution) – The stage to add

Return type:

None

abstractmethod remove_stage(execution, stage_id)[source]

Remove a stage from an execution.

Parameters:

  • execution (Workflow) – The execution

  • stage_id (str) – The stage ID to remove

Return type:

None

abstractmethod retrieve_stage(stage_id)[source]

Retrieve a single stage by ID.

The returned stage will have a partial parent execution attached (containing metadata but no other stages).

Parameters:

stage_id (str) – The stage ID

Returns:

The stage execution

Raises:

ValueError – If stage not found

Return type:

StageExecution

abstractmethod get_upstream_stages(execution_id, stage_ref_id)[source]

Get upstream stages for a given stage.

Parameters:

  • execution_id (str) – The execution ID

  • stage_ref_id (str) – The reference ID of the stage

Returns:

List of upstream stages

Return type:

list[StageExecution]

abstractmethod get_downstream_stages(execution_id, stage_ref_id)[source]

Get downstream stages for a given stage.

Parameters:

  • execution_id (str) – The execution ID

  • stage_ref_id (str) – The reference ID of the stage

Returns:

List of downstream stages

Return type:

list[StageExecution]

abstractmethod get_synthetic_stages(execution_id, parent_stage_id)[source]

Get synthetic stages for a given parent stage.

Parameters:

  • execution_id (str) – The execution ID

  • parent_stage_id (str) – The parent stage ID

Returns:

List of synthetic stages

Return type:

list[StageExecution]

abstractmethod get_merged_ancestor_outputs(execution_id, stage_ref_id)[source]

Get merged outputs from all ancestor stages.

Traverses the DAG upstream, collects outputs, and merges them according to topological order (latest wins).

Parameters:

  • execution_id (str) – The execution ID

  • stage_ref_id (str) – The reference ID of the stage

Returns:

Merged dictionary of outputs

Return type:

dict[str, Any]

abstractmethod retrieve_by_pipeline_config_id(pipeline_config_id, criteria=None)[source]

Retrieve executions by pipeline config ID.

Parameters:

  • pipeline_config_id (str) – The pipeline config ID

  • criteria (WorkflowCriteria | None) – Optional query criteria

Returns:

Iterator of matching executions

Return type:

Iterator[Workflow]

abstractmethod retrieve_by_application(application, criteria=None)[source]

Retrieve executions by application.

Parameters:

  • application (str) – The application name

  • criteria (WorkflowCriteria | None) – Optional query criteria

Returns:

Iterator of matching executions

Return type:

Iterator[Workflow]

abstractmethod pause(execution_id, paused_by)[source]

Pause an execution.

Parameters:

  • execution_id (str) – The execution ID

  • paused_by (str) – Who paused it

Return type:

None

abstractmethod resume(execution_id)[source]

Resume a paused execution.

Parameters:

execution_id (str) – The execution ID

Return type:

None

abstractmethod cancel(execution_id, canceled_by, reason)[source]

Cancel an execution.

Parameters:

  • execution_id (str) – The execution ID

  • canceled_by (str) – Who canceled it

  • reason (str) – Cancellation reason

Return type:

None

is_message_processed(message_id)[source]

Check if a message has already been processed.

Used for idempotency - prevents duplicate message processing.

Parameters:

message_id (str) – The unique message ID

Returns:

True if the message has been processed before

Return type:

bool

mark_message_processed(message_id, handler_type=None, execution_id=None)[source]

Mark a message as successfully processed.

Parameters:

  • message_id (str) – The unique message ID

  • handler_type (str | None) – Optional handler type for debugging

  • execution_id (str | None) – Optional execution ID for debugging

Return type:

None

cleanup_old_processed_messages(max_age_hours=24.0)[source]

Clean up old processed message records.

Parameters:

max_age_hours (float) – Delete records older than this many hours

Returns:

Number of records deleted

Return type:

int

get_processed_message_ids(limit=None)[source]

Return processed message IDs, for hydrating an in-memory dedup cache.

Parameters:

limit (int | None) – Maximum number of IDs to return (None for all)

Returns:

The message IDs, or None when the store cannot enumerate them. Callers receiving None must treat any in-memory dedup cache as non-authoritative and always confirm with is_message_processed().

Return type:

list[str] | None

is_healthy()[source]

Check if the repository is healthy.

Returns:

True if healthy

Return type:

bool

count_by_application(application)[source]

Count executions for an application.

Parameters:

application (str) – The application name

Returns:

Number of executions

Return type:

int

exists(execution_id)[source]

Check if a workflow exists.

Parameters:

execution_id (str) – The workflow ID to check

Returns:

True if the workflow exists, False otherwise

Return type:

bool

transaction(queue=None)[source]

Create an atomic transaction for store + queue operations.

Use this when you need to atomically update both stage state AND queue a message. This prevents orphaned workflows from crashes between separate store and queue operations.

Default implementation provides a no-op transaction that just delegates to the normal store methods. SQLite and PostgreSQL implementations provide true atomic transactions.

Parameters:

queue (Queue | None) – Optional queue for pushing messages. Required for backends that don’t have integrated queue support.

Return type:

Iterator[StoreTransaction]

Usage:
with store.transaction(queue) as txn:

txn.store_stage(stage) txn.push_message(message)

# Commits on success, rolls back on exception

Yields:

StoreTransaction with store_stage() and push_message() methods

Parameters:

queue (Queue | None)

Return type:

Iterator[StoreTransaction]

class stabilize.persistence.store.WorkflowCriteria(page_size=20, statuses=None, start_time_before=None, start_time_after=None)[source]

Bases: object

Criteria for querying executions.

Parameters:

  • page_size (int)

  • statuses (set[WorkflowStatus] | None)

  • start_time_before (int | None)

  • start_time_after (int | None)

page_size: int = 20
statuses: set[WorkflowStatus] | None = None
start_time_before: int | None = None
start_time_after: int | None = None
exception stabilize.persistence.store.WorkflowNotFoundError(execution_id)[source]

Bases: Exception

Raised when an execution cannot be found.

Parameters:

execution_id (str)

class stabilize.persistence.store.StoreTransaction[source]

Bases: ABC

Abstract interface for atomic store + queue transactions.

Implementations must ensure that store_stage() and push_message() are committed atomically - either both succeed or both are rolled back.

property is_atomic: bool

Whether this transaction provides true database-level atomicity.

Returns True if all operations within this transaction are guaranteed to either all succeed or all fail atomically (no partial state).

Override this in subclasses. Default is False for safety.

abstractmethod store_stage(stage, expected_phase=None)[source]

Store or update a stage within the transaction.

Parameters:

  • stage (StageExecution) – The stage to store

  • expected_phase (str | None) – If provided, adds status check to WHERE clause for phase-aware optimistic locking (CAS pattern).

Return type:

None

abstractmethod update_workflow_status(workflow)[source]

Update workflow status within the transaction.

Parameters:

workflow (Workflow)

Return type:

None

abstractmethod push_message(message, delay=0)[source]

Push a message to the queue within the transaction.

Parameters:
Return type:

None

abstractmethod mark_message_processed(message_id, handler_type=None, execution_id=None)[source]

Mark a message as successfully processed within the transaction.

This ensures that message processing is only marked complete if the transaction commits successfully.

Parameters:

  • message_id (str)

  • handler_type (str | None)

  • execution_id (str | None)

Return type:

None

acquire_claim(execution_id, claim_key, stage_id, steal_if_owner_terminal=False)[source]

Atomically acquire a named claim for a stage within the transaction.

Used to serialize sibling stages competing for the same resource — mutex keys (WCP-17/39/40) and deferred-choice groups (WCP-16) — via a unique (execution_id, claim_key) row. The per-stage-row CAS cannot do this: two DIFFERENT stages each pass their own row’s CAS.

Parameters:

  • execution_id (str) – The workflow execution

  • claim_key (str) – Namespaced key, e.g. “mutex:<key>” or “choice:<group>”

  • stage_id (str) – The stage attempting to acquire

  • steal_if_owner_terminal (bool) – Transfer the claim when the current owner stage has reached a terminal status (mutex release semantics). Leave False for claims that are decided once (deferred choice).

Returns:

True if this stage holds the claim (newly acquired, already owned, or transferred); False if another live stage holds it.

Return type:

bool

Note

The default implementation returns True (no enforcement) so custom store transactions without claim support keep their existing behavior.

class stabilize.persistence.store.NoOpTransaction(store, queue=None)[source]

Bases: StoreTransaction

Default transaction that buffers operations and flushes on commit.

WARNING: This implementation is NOT truly atomic. A crash during flush can leave partial state in the database. This is only suitable for: - Testing environments - Non-critical workloads where eventual consistency is acceptable

For production critical systems requiring 100% atomicity, use: - SqliteWorkflowStore (provides true atomic transactions) - PostgresWorkflowStore (provides true atomic transactions)

Operations are buffered and flushed in careful order to minimize inconsistency windows, but true atomicity is impossible without database-level transaction support.

Parameters:
store_stage(stage, expected_phase=None)[source]

Buffer stage to be stored when transaction completes.

Stages are buffered and flushed when the context manager exits successfully. If an exception occurs, stages are not stored.

Parameters:

  • stage (StageExecution) – Stage to store

  • expected_phase (str | None) – If provided, passed to store.store_stage() for CAS.

Return type:

None

update_workflow_status(workflow)[source]

Buffer workflow status update when transaction completes.

Workflows are buffered and flushed when the context manager exits successfully. If an exception occurs, workflows are not updated.

Parameters:

workflow (Workflow)

Return type:

None

push_message(message, delay=0)[source]

Buffer message to be pushed when transaction completes.

Messages are buffered and flushed when the context manager exits successfully. If an exception occurs, messages are not pushed.

Parameters:
Return type:

None

mark_message_processed(message_id, handler_type=None, execution_id=None)[source]

Buffer processed message mark to be stored when transaction completes.

Parameters:

  • message_id (str)

  • handler_type (str | None)

  • execution_id (str | None)

Return type:

None

SQLite persistence implementation.

class stabilize.persistence.sqlite.SqliteWorkflowStore(connection_string, create_tables=False)[source]

Bases: SqliteWorkflowCrudMixin, SqliteStageOpsMixin, SqliteQueriesMixin, WorkflowStore

SQLite implementation of WorkflowStore.

Uses native sqlite3 for file-based or in-memory storage. Suitable for development, testing, and single-node deployments.

Features: - WAL mode for better concurrent read performance - Foreign key support enabled - JSON stored as TEXT strings - Arrays stored as JSON strings - Thread-local connections managed by singleton ConnectionManager

Parameters:

  • connection_string (str)

  • create_tables (bool)

close()[source]

Close SQLite connection for current thread.

Return type:

None

is_healthy()[source]

Check if the database connection is healthy.

Return type:

bool

transaction(queue=None)[source]

Create atomic transaction for store + queue operations.

Use this when you need to atomically update both stage state AND queue a message. This prevents orphaned workflows from crashes between separate store and queue operations.

SQLite implementation writes directly to the queue_messages table in the same transaction, so the queue parameter is ignored.

Parameters:

queue (Queue | None) – Ignored (for API compatibility with base class)

Return type:

Iterator[StoreTransaction]

Usage:
with store.transaction() as txn:

txn.store_stage(stage) txn.push_message(message)

# Auto-commits on success, rolls back on exception

Yields:

AtomicTransaction with store_stage() and push_message() methods

Parameters:

queue (Queue | None)

Return type:

Iterator[StoreTransaction]

is_message_processed(message_id)[source]

Check if a message has already been processed.

Parameters:

message_id (str)

Return type:

bool

mark_message_processed(message_id, handler_type=None, execution_id=None)[source]

Mark a message as successfully processed.

Parameters:

  • message_id (str)

  • handler_type (str | None)

  • execution_id (str | None)

Return type:

None

cleanup_old_processed_messages(max_age_hours=24.0)[source]

Clean up old processed message records.

Parameters:

max_age_hours (float)

Return type:

int

get_processed_message_ids(limit=None)[source]

Return processed message IDs, for hydrating an in-memory dedup cache.

Parameters:

limit (int | None)

Return type:

list[str] | None

PostgreSQL persistence implementation.

class stabilize.persistence.postgres.PostgresWorkflowStore(connection_string)[source]

Bases: WorkflowStore

PostgreSQL implementation of WorkflowStore.

Uses native psycopg3 with connection pooling for database operations. Supports concurrent access and provides efficient queries for pipeline execution tracking.

Parameters:

connection_string (str)

close()[source]

Close the connection pool via connection manager.

Return type:

None

store(execution)[source]

Store a complete execution.

Parameters:

execution (Workflow)

Return type:

None

retrieve(execution_id)[source]

Retrieve an execution by ID.

Parameters:

execution_id (str)

Return type:

Workflow

retrieve_execution_summary(execution_id)[source]

Retrieve execution metadata without stages.

Parameters:

execution_id (str)

Return type:

Workflow

update_status(execution)[source]

Update execution status.

Parameters:

execution (Workflow)

Return type:

None

delete(execution_id)[source]

Delete an execution.

Parameters:

execution_id (str)

Return type:

None

store_stage(stage, expected_phase=None, connection=None)[source]

Store or update a stage.

Parameters:

  • stage (Any) – The stage to store

  • expected_phase (str | None) – If provided, adds status check to WHERE clause for phase-aware optimistic locking.

  • connection (Any | None) – Optional existing connection to use

Return type:

None

add_stage(stage)[source]

Add a new stage.

Parameters:

stage (Any)

Return type:

None

remove_stage(execution, stage_id)[source]

Remove a stage.

Parameters:
Return type:

None

retrieve_stage(stage_id)[source]

Retrieve a single stage by ID.

Parameters:

stage_id (str)

Return type:

Any

get_upstream_stages(execution_id, stage_ref_id)[source]

Get upstream stages with tasks loaded.

Parameters:

  • execution_id (str)

  • stage_ref_id (str)

Return type:

list[Any]

get_downstream_stages(execution_id, stage_ref_id)[source]

Get downstream stages with tasks loaded.

Parameters:

  • execution_id (str)

  • stage_ref_id (str)

Return type:

list[Any]

get_synthetic_stages(execution_id, parent_stage_id)[source]

Get synthetic stages with tasks loaded.

Parameters:

  • execution_id (str)

  • parent_stage_id (str)

Return type:

list[Any]

get_merged_ancestor_outputs(execution_id, stage_ref_id)[source]

Get merged outputs from all ancestor stages.

Parameters:

  • execution_id (str)

  • stage_ref_id (str)

Return type:

dict[str, Any]

retrieve_by_pipeline_config_id(pipeline_config_id, criteria=None)[source]

Retrieve executions by pipeline config ID.

Parameters:
Return type:

Iterator[Workflow]

retrieve_by_application(application, criteria=None)[source]

Retrieve executions by application.

Parameters:
Return type:

Iterator[Workflow]

get_all_pending_workflows(criteria=None)[source]

Get all workflows matching criteria across all applications (for recovery).

Parameters:

criteria (WorkflowCriteria | None)

Return type:

Iterator[Workflow]

pause(execution_id, paused_by)[source]

Pause an execution.

Parameters:

  • execution_id (str)

  • paused_by (str)

Return type:

None

resume(execution_id)[source]

Resume a paused execution.

Parameters:

execution_id (str)

Return type:

None

cancel(execution_id, canceled_by, reason)[source]

Cancel an execution.

Parameters:

  • execution_id (str)

  • canceled_by (str)

  • reason (str)

Return type:

None

is_message_processed(message_id)[source]

Check if a message has already been processed.

Parameters:

message_id (str)

Return type:

bool

mark_message_processed(message_id, handler_type=None, execution_id=None)[source]

Mark a message as successfully processed.

Parameters:

  • message_id (str)

  • handler_type (str | None)

  • execution_id (str | None)

Return type:

None

cleanup_old_processed_messages(max_age_hours=24.0)[source]

Clean up old processed message records.

Parameters:

max_age_hours (float)

Return type:

int

get_processed_message_ids(limit=None)[source]

Return processed message IDs, for hydrating an in-memory dedup cache.

Parameters:

limit (int | None)

Return type:

list[str] | None

is_healthy()[source]

Check if the database connection is healthy.

Return type:

bool

transaction(queue=None)[source]

Create an atomic transaction for store + queue operations.

Parameters:

queue (Any | None)

Return type:

Iterator[StoreTransaction]