perceptic-workflow-sdk 0.50.94

Python SDK for Perceptic Workflow definitions - cross-language Temporal workflow types

Perceptic Workflow SDK

Python SDK for Perceptic Workflow definitions. This package provides cross-language Temporal workflow types that are compatible with the Java workflow definitions in perceptic-core-client.

Installation

pip install perceptic-workflow-sdk

Usage

Workflow Events

from perceptic_workflow import (
    WorkflowEvent,
    InfoEvent,
    CheckpointEvent,
    UserInputEvent,
    UserInputRequestEvent,
    WorkflowCheckpointStatus,
)

# Create events
info_event = InfoEvent(
    event_id=1,
    type="progress",
    payload={"step": "processing"},
    timestamp=datetime.now()
)

Implementing Workflows

The SDK provides a mixin class for implementing Perceptic-compatible workflows:

from temporalio import workflow
from perceptic_workflow import PercepticWorkflowMixin, WorkflowEvent

@workflow.defn
class MyWorkflow(PercepticWorkflowMixin):
    def __init__(self):
        self._events: list[WorkflowEvent] = []
        self._paused = False

    @workflow.run
    async def run(self, input: dict) -> dict:
        # Your workflow logic here
        pass

    @workflow.update(name=PercepticWorkflowMixin.UPDATE_SUBMIT_USER_INPUT)
    async def submit_user_input(self, inputs: dict) -> None:
        self._paused = False
        # Handle user input

    @workflow.update(name=PercepticWorkflowMixin.UPDATE_INTERRUPT)
    async def interrupt(self, reason: str) -> None:
        # Handle interruption
        pass

    @workflow.query(name=PercepticWorkflowMixin.QUERY_IS_PAUSED)
    def is_paused(self) -> bool:
        return self._paused

    @workflow.query(name=PercepticWorkflowMixin.QUERY_GET_EVENTS)
    def get_events(self, after_event_id: int | None = None) -> list[WorkflowEvent]:
        if after_event_id is None:
            return self._events
        return [e for e in self._events if e.event_id > after_event_id]

Compatibility

This package is designed to be compatible with:

• perceptic-core-client (Java)
• perceptic-core-workflow-runtime (Java)

The workflow definitions and event types are generated from the same JSON Schema sources to ensure cross-language compatibility.