Workflow Decorator

The @workflow decorator marks a function as a Tray.io workflow and carries its configuration.

Basic usage

from riffle.sdk import connectors, workflow

@workflow(title="Sync Contacts")
def sync_contacts(trigger: connectors.callable_trigger.trigger):
    query = connectors.salesforce.find_records(
        object="Contact",
    )

The function’s single parameter is the trigger step. Its type annotation (connectors.callable_trigger.trigger) specifies which connector and operation acts as the workflow’s entry point.

Parameters

Parameter	Type	Description
title	str	(required) Workflow title as shown in the Tray.io UI.
enabled	bool	Whether the workflow is enabled. Omit to leave unchanged.
tags	list[str]	Tags applied to the workflow for organization.
input_schema	Schema	Reference to a Schema class defining the workflow’s input. Passed as the class itself, not a string.
output_schema	Schema	Reference to a Schema class defining the workflow’s output.
alerting	str	API name of another workflow to call when this workflow errors.
trigger_auth	str	Title of the authentication credential used by the trigger.
trigger_config	dict	Configuration properties passed to the trigger connector.

Examples

With schemas

from riffle.sdk import Field, Schema, connectors, workflow

class WebhookPayload(Schema):
    event_type: str
    record_id: str = Field(title="Record ID")

class WebhookResult(Schema):
    success: bool
    message: str

@workflow(
    title="Handle Webhook",
    enabled=True,
    tags=["integration", "salesforce"],
    input_schema=WebhookPayload,
    output_schema=WebhookResult,
)
def handle_webhook(trigger: connectors.callable_trigger.trigger):
    update = connectors.salesforce.update_record(
        object_id=trigger.record_id,
    )

With trigger configuration

@workflow(
    title="Scheduled Sync",
    trigger_config={
        "schedule": "0 9 * * 1-5",
    },
)
def scheduled_sync(trigger: connectors.scheduled_trigger.trigger):
    records = connectors.salesforce.find_records(
        object="Account",
    )

With alerting

@workflow(
    title="Critical Pipeline",
    alerting="error-handler",
)
def critical_pipeline(trigger: connectors.callable_trigger.trigger):
    result = connectors.script.execute(
        code=trigger.payload,
    )

The alerting value is the API name (filename without .py) of another workflow in the same project.