Skip to main content

WorkflowProtocol

Protocol for workflow state management.

Implementations must handle:
- Starting workflows on documents
- Tracking current workflow state
- Validating and executing state transitions
- Checking available actions based on user permissions

Thread Safety:
    Implementations must be thread-safe. Multiple concurrent transitions
    on the same document should be handled correctly (e.g., using locks
    or optimistic concurrency control).

Database Compatibility:
    Implementations must work with both SQLite (testing) and PostgreSQL
    (production). Use SQLAlchemy's database-agnostic features.

Source: workflow.py

Methods​

start_workflow​

async def start_workflow(self,
    doctype: str,
    doc_id: str,
    workflow_name: str,
    user: UserContext,
) -> WorkflowStateInfo

Start a workflow on a document.

    Creates initial workflow state record. If workflow already exists,
    this operation is idempotent and returns current state.

    Args:
        doctype: The DocType name
        doc_id: The document ID
        workflow_name: Name of the workflow to start
        user: User context starting the workflow

    Returns:
        WorkflowStateInfo with initial state

    Raises:
        ValueError: If workflow_name is invalid for this doctype

get_workflow_state​

async def get_workflow_state(self, doctype: str, doc_id: str
) -> WorkflowStateInfo | None

Get current workflow state for a document.

    Args:
        doctype: The DocType name
        doc_id: The document ID

    Returns:
        Current workflow state info, or None if no workflow active

transition​

async def transition(self, request: TransitionRequest) -> TransitionResult

Attempt to transition a document to a new workflow state.

    Validates:
    - Workflow exists for document
    - Transition action is valid from current state
    - User has required role for this transition
    - Any custom validation hooks pass

    Args:
        request: Transition request with action and user context

    Returns:
        TransitionResult indicating success/failure and new state

get_available_actions​

async def get_available_actions(self, doctype: str, doc_id: str, user: UserContext
) -> list[str]

Get list of actions available to user in current state.

    Filters actions based on:
    - Current workflow state
    - User's roles
    - Any custom permission logic

    Args:
        doctype: The DocType name
        doc_id: The document ID
        user: User context to check permissions for

    Returns:
        List of action names user can perform