​
Overview

Events in Sample workflows allow you to trigger workflows and control their execution based on external or internal events. This creates a reactive workflow system where workflows can respond to changes in your application state.

Events functionality requires samplehc >= 0.3.0 and workflows-py >= 0.1.18.

​
Emitting Events

You can emit events from your workflow functions using the SampleHC client. Events consist of a name and an optional payload.

​
Basic Event Emission

from client_manager import SampleHealthcareClient

def emit_order_updated(ctx: WorkflowRunContext):
    """Handle the update action - triggers eligibility if status is new"""
    order = get_order(ctx)
    client = get_client(ctx)
    
    client.v2.events.emit(
        name="order-updated",
        payload={
            "order_id": order["order_id"], 
            "bill_type": get_bill_type(ctx)
        },
    )

​
Event Structure

Events have the following structure:

  • name: A string identifier for the event type (e.g., “order-updated”, “order-created”)
  • payload: Any JSON-serializable data relevant to the event

​
Listening to Events

Workflows can listen to events using two mechanisms: start_on and cancel_on.

​
Starting Workflows on Events

Use start_on to automatically start a workflow when specific events are received:

from workflows_py.workflow import Workflow

workflow = Workflow(
    get_display_col_values={"Order ID": lambda ctx: ctx.get_start_data()["order_id"]},
    start_on={
        "order-created": lambda ctx: create_insurance_verification_start_data_from_start_on_ctx(ctx),
        "order-updated": lambda ctx: create_insurance_verification_start_data_from_start_on_ctx(ctx),
    },
)

​
Start Handler Return Values

The start_on handler function can return:

  • An object with optional start_data field: Starts the workflow with the provided start data
  • None: Does not start the workflow
def handle_order_created(ctx):
    # Start the workflow with custom data
    return {
        "start_data": {
            "order_id": ctx.event.payload["order_id"],
            "priority": "high"
        }
    }

def handle_low_priority_order(ctx):
    # Don't start workflow for low priority orders
    if ctx.event.payload.get("priority") == "low":
        return None
    
    # Start workflow for other priorities
    return {"start_data": {"order_id": ctx.event.payload["order_id"]}}

# These handlers would be used in your workflow configuration like this:
workflow = Workflow(
    get_display_col_values={"Order ID": lambda ctx: ctx.get_start_data()["order_id"]},
    start_on={
        "order-created": handle_order_created,
        "order-updated": handle_low_priority_order,
    },
)

​
Canceling Workflows on Events

Use cancel_on to automatically cancel running workflows when specific events are received:

workflow = Workflow(
    # ... other configuration
    cancel_on={
        "order-updated": lambda ctx: ctx.get_start_data()["order_id"] == ctx.event.payload.get("order_id"),
    },
)

​
Cancel Handler Return Values

The cancel_on handler function should return:

  • True: Cancel the workflow run
  • False: Keep the workflow running
def should_cancel_on_order_update(ctx):
    # Cancel if this workflow is processing the same order
    workflow_order_id = ctx.get_start_data()["order_id"]
    event_order_id = ctx.event.payload.get("order_id")
    
    return workflow_order_id == event_order_id

​
Event Context

When handling events, you have access to an event context that provides:

​
StartOnContext

Available in start_on handlers:

def handle_start_event(ctx: StartOnContext):
    # Access the event that triggered this handler
    event_name = ctx.event.name
    event_payload = ctx.event.payload
    
    # Access authentication token (for API calls)
    backend_token = ctx.backend_token
    
    # Return start data or None
    return {"start_data": {"triggered_by": event_name}}

​
CancelOnContext

Available in cancel_on handlers:

def handle_cancel_event(ctx: CancelOnContext):
    # Access the event that triggered this handler
    event_name = ctx.event.name
    event_payload = ctx.event.payload
    
    # Access the workflow's original start data
    start_data = ctx.get_start_data()
    
    # Return boolean decision
    return should_cancel_workflow(start_data, event_payload)

​
Common Patterns

​
Order Processing Workflow

workflow = Workflow(
    get_display_col_values={
        "Order ID": lambda ctx: ctx.get_start_data()["order_id"],
        "Status": lambda ctx: ctx.get_start_data().get("status", "pending")
    },
    start_on={
        "order-created": lambda ctx: {
            "start_data": {
                "order_id": ctx.event.payload["order_id"],
                "status": "processing"
            }
        }
    },
    cancel_on={
        "order-cancelled": lambda ctx: (
            ctx.get_start_data()["order_id"] == ctx.event.payload["order_id"]
        )
    }
)

​
Multi-Event Workflow

workflow = Workflow(
    start_on={
        "patient-admitted": lambda ctx: create_admission_workflow_data(ctx),
        "insurance-verified": lambda ctx: create_insurance_workflow_data(ctx),
        "urgent-case": lambda ctx: create_urgent_workflow_data(ctx),
    },
    cancel_on={
        "patient-discharged": lambda ctx: (
            ctx.get_start_data()["patient_id"] == ctx.event.payload["patient_id"]
        ),
        "case-closed": lambda ctx: (
            ctx.get_start_data()["case_id"] == ctx.event.payload["case_id"]
        ),
    }
)

​
Best Practices

​
Event Naming

  • Use descriptive, consistent event names (e.g., “order-created”, “patient-admitted”)
  • Include entity type and action (e.g., “user-registered”, “document-processed”)
  • Use kebab-case for event names

​
Payload Design

  • Include essential identifiers in the payload (IDs, timestamps)
  • Keep payloads lightweight - avoid large objects
  • Include context needed for decision making

​
Error Handling

def safe_event_handler(ctx):
    try:
        # Your event handling logic
        return {"start_data": process_event(ctx.event)}
    except Exception as e:
        # Log error and decide whether to start workflow
        logger.error(f"Error processing event: {e}")
        return None  # Don't start workflow on error

​
Performance Considerations

  • Keep event handlers lightweight and fast
  • Avoid heavy computations in start_on and cancel_on handlers
  • Use specific event matching to avoid unnecessary workflow starts/cancellations

​
Troubleshooting

​
Common Issues

Workflow not starting on events:

  • Verify event names match exactly between emit and listen
  • Check that handler functions return appropriate values
  • Ensure version requirements are met

Unexpected workflow cancellations:

  • Review cancel_on handler logic
  • Check event payload structure
  • Verify cancellation conditions are specific enough

Missing event data:

  • Ensure event payload contains required fields
  • Check that get_start_data() returns expected structure
  • Verify event emission includes necessary context