cjm-workflow-state

Install

pip install cjm_workflow_state

Project Structure

nbs/
├── history.ipynb     # Generic undo history stack for workflow operations.
└── state_store.ipynb # SQLite-backed workflow state storage for persistence across restarts.

Total: 2 notebooks

Module Dependencies

graph LR
    history[history<br/>History]
    state_store[state_store<br/>State Store]

No cross-module dependencies detected.

CLI Reference

No CLI commands found in this project.

Module Overview

Detailed documentation for each module in the project:

History (history.ipynb)

Generic undo history stack for workflow operations.

Import

from cjm_workflow_state.history import (
    DEFAULT_MAX_HISTORY_DEPTH,
    push_history,
    pop_history
)

Functions

def push_history(
    history: List[Dict[str, Any]],  # Current history stack
    snapshot: Dict[str, Any],  # State snapshot to save
    max_depth: int = DEFAULT_MAX_HISTORY_DEPTH,  # Maximum history depth
) -> List[Dict[str, Any]]:  # Updated history stack
    "Push a state snapshot onto the history stack, enforcing max depth."

def pop_history(
    history: List[Dict[str, Any]],  # Current history stack
) -> Optional[Tuple[Dict[str, Any], List[Dict[str, Any]]]]:  # (snapshot, updated_history) or None
    "Pop the most recent snapshot from the history stack."

Variables

DEFAULT_MAX_HISTORY_DEPTH = 50

State Store (state_store.ipynb)

SQLite-backed workflow state storage for persistence across restarts.

Import

from cjm_workflow_state.state_store import (
    SessionSummary,
    SQLiteWorkflowStateStore
)

Classes

@dataclass
class SessionSummary:
    "Lightweight metadata for a single workflow session."
    
    flow_id: str  # Workflow identifier
    session_id: str  # Session identifier
    label: Optional[str]  # Human-readable label, or None to defer to a generator
    current_step: Optional[str]  # Current StepFlow step ID, or None
    created_at: str  # ISO timestamp string from SQLite CURRENT_TIMESTAMP
    updated_at: str  # ISO timestamp string from SQLite CURRENT_TIMESTAMP
    state_size_bytes: int  # Length of state_json column — cheap size hint

class SQLiteWorkflowStateStore:
    def __init__(
        self,
        db_path:Path # Path to SQLite database file
    )
    "SQLite-backed workflow state storage for persistence across restarts."
    
    def __init__(
            self,
            db_path:Path # Path to SQLite database file
        )
        "Initialize the state store and create or migrate the schema."
    
    def get_current_step(
            self,
            flow_id:str, # Workflow identifier
            session_id:str # Session identifier string
        ) -> Optional[str]: # Current step ID, or None if no row exists
        "Get the current step ID for a session."
    
    def set_current_step(
            self,
            flow_id:str, # Workflow identifier
            session_id:str, # Session identifier string
            step_id:str # Step ID to set as current
        ) -> None
        "Set the current step ID for a session, upserting the row if needed."
    
    def get_state(
            self,
            flow_id:str, # Workflow identifier
            session_id:str # Session identifier string
        ) -> Dict[str, Any]: # Full session state dictionary
        "Get the full state dictionary for a session."
    
    def update_state(
            self,
            flow_id:str, # Workflow identifier
            session_id:str, # Session identifier string
            updates:Dict[str, Any] # Top-level state keys to merge in
        ) -> None
        "Merge updates into the session state, upserting the row if needed."
    
    def clear_state(
            self,
            flow_id:str, # Workflow identifier
            session_id:str # Session identifier string
        ) -> None
        "Delete the session row entirely. Idempotent."
    
    def get_step_state(
            self,
            flow_id:str, # Workflow identifier
            session_id:str, # Session identifier string
            step_id:str # Step identifier
        ) -> Dict[str, Any]: # State dictionary for that step
        "Get the state dictionary for a specific step."
    
    def update_step_state(
            self,
            flow_id:str, # Workflow identifier
            session_id:str, # Session identifier string
            step_id:str, # Step identifier
            updates:Dict[str, Any] # Step state keys to merge in
        ) -> None
        "Merge updates into a specific step's state."
    
    def clear_step_state(
            self,
            flow_id:str, # Workflow identifier
            session_id:str, # Session identifier string
            step_id:str # Step identifier
        ) -> None
        "Remove a specific step's state while preserving the rest of the session."
    
    def create_session(
            self,
            flow_id:str, # Workflow identifier
            session_id:Optional[str]=None, # Pre-chosen session ID (auto-generated UUID4 if None)
            label:Optional[str]=None # Optional human-readable label
        ) -> str: # The created session_id
        "Insert a new empty session row and return its session_id."
    
    def session_exists(
            self,
            flow_id:str, # Workflow identifier
            session_id:str # Session identifier string
        ) -> bool: # True if a row exists for the given session
        "Check whether a session row exists."
    
    def get_session_summary(
            self,
            flow_id:str, # Workflow identifier
            session_id:str # Session identifier string
        ) -> Optional[SessionSummary]: # Session metadata, or None if not found
        "Return lightweight metadata for a single session."
    
    def list_sessions(
            self,
            flow_id:str, # Workflow identifier to list sessions for
            order_by:str="updated_at", # Sort column: "updated_at", "created_at", or "label"
            descending:bool=True # Sort direction
        ) -> List[SessionSummary]: # All sessions for this flow, ordered
        "List all sessions for a flow as lightweight SessionSummary records."
    
    def set_session_label(
            self,
            flow_id:str, # Workflow identifier
            session_id:str, # Session identifier string
            label:Optional[str] # New label, or None to clear
        ) -> None
        "Update the session label. No-op if the session does not exist."
    
    def delete_session(
            self,
            flow_id:str, # Workflow identifier
            session_id:str # Session identifier string
        ) -> None
        "Delete a session row. Alias of `clear_state` for symmetry with the session API."