Event-triggered workflows

from polos import workflow, WorkflowContext, EventPayload

@workflow(
    id="user-signup-handler",
    trigger_on_event="user/signup"
)
async def user_signup_handler(ctx: WorkflowContext, payload: EventPayload):
    # Triggered when event published to "user.signup" topic
    user_id = payload.data["user_id"]
    email = payload.data["email"]

    # Send welcome email
    await ctx.step.run("send_welcome", send_welcome_email, user_id, email)

    # Create sample data
    await ctx.step.run("setup_account", create_sample_data, user_id)

    return {"status": "onboarded", "user_id": user_id}

@workflow(
    id="notification-handler",
    trigger_on_event="notifications/new"
)
async def notification_handler(ctx: WorkflowContext, payload: EventPayload):
    # Single event
    print(f"Event ID: {payload.id}")
    print(f"Sequence ID: {payload.sequence_id}")
    print(f"Topic: {payload.topic}")
    print(f"Event Type: {payload.event_type}")
    print(f"Data: {payload.data}")  # This is a dict
    print(f"Created At: {payload.created_at}")

    # Process the event
    notification_data = payload.data
    await ctx.step.run("send", send_notification, notification_data)

@workflow
async def create_user(ctx: WorkflowContext, input: CreateUserInput):
    # Create user
    user = await ctx.step.run("create", create_user_record, input)

    # Publish event (triggers event-triggered workflows)
    await ctx.step.publish_event(
        "publish_signup",
        topic="user/signup",
        data={
            "user_id": user.id,
            "email": user.email,
            "name": user.name
        },
        event_type="user_created"
    )

    return {"user_id": user.id}

import httpx

async def publish_event(topic: str, data: dict):
    """Publish an event via API."""
    async with httpx.AsyncClient() as client:
        response = await client.post(
            "https://api.polos.ai/api/v1/events/publish",
            headers={
                "Authorization": "Bearer YOUR_API_KEY",
                "Content-Type": "application/json"
            },
            json={
                "topic": topic,
                "events": [{
                    "data": data,
                    "event_type": "custom_event"
                }]
            }
        )
        response.raise_for_status()

# Trigger workflows listening to "order.completed"
await publish_event(
    "order.completed",
    {"order_id": "ord_123", "total": 99.99}
)

@workflow(
    id="batch-processor",
    trigger_on_event="analytics/events",
    batch_size=10,              # Process up to 10 events
    batch_timeout_seconds=30    # Or wait max 30 seconds
)
async def batch_processor(ctx: WorkflowContext, payload: BatchEventPayload):
    # Multiple events in the payload
    events = payload.events

    print(f"Processing batch of {len(events)} events")

    # Extract all event data
    analytics_data = [event.data for event in events]

    # Process batch
    await ctx.step.run("batch_insert", insert_analytics, analytics_data)

    return {"processed": len(events)}

@workflow(
    id="immediate-handler",
    trigger_on_event="order/created"
)
async def immediate_handler(ctx: WorkflowContext, payload: EventPayload):
    """Process each order immediately."""
    order_id = payload.data["order_id"]

    await ctx.step.run("send_confirmation", send_order_confirmation, order_id)
    return {"handler": "immediate"}

@workflow(
    id="batched-handler",
    trigger_on_event="order/created",
    batch_size=5,
    batch_timeout_seconds=60
)
async def batched_handler(ctx: WorkflowContext, payload: BatchEventPayload):
    """Process orders in batches for analytics."""
    events = payload.events

    order_ids = [e.data["order_id"] for e in events]
    await ctx.step.run("batch_analytics", update_analytics, order_ids)

    return {"handler": "batched", "count": len(events)}

@workflow(
    id="high-priority-handler",
    trigger_on_event="notifications/system"
)
async def high_priority_handler(ctx: WorkflowContext, payload: PayloadEvent):
    # Only process high priority notifications
    if payload.event_type == "high_priority":
        await ctx.step.run("alert", send_alert, payload.data)
    else:
        print(f"Skipping event type: {payload.event_type}")

# Publish to specific topics
await ctx.step.publish_event(
    "publish",
    topic="notifications/high_priority",  # Specific topic
    data=notification_data
)

@workflow(
    id="high-priority-handler",
    trigger_on_event="notifications/high_priority"  # Only high priority
)
async def high_priority_handler(ctx: WorkflowContext, payload: EventPayload):
    # All events on this topic are high priority
    await ctx.step.run("alert", send_alert, payload.data)