Streaming events

from polos import workflow, WorkflowContext, PolosClient, events

@workflow
async def data_pipeline(ctx: WorkflowContext, input: PipelineInput):
    data = await ctx.step.run("extract", extract_data, input)
    transformed = await ctx.step.run("transform", transform_data, data)
    final = await ctx.step.invoke_and_wait("child_pipeline", child_pipeline, transformed)
    await ctx.step.run("load", load_data, final)

# Start workflow
client = PolosClient()
handle = await data_pipeline.invoke(
    client,
    PipeInput(source="database", table="orders")
)

# Stream events
async for event in events.stream_workflow(client, handle.root_workflow_id, handle.id):
    if event.event_type == "step_start":
        print(f"Step started: {event.data.get('step_key')}")

    elif event.event_type == "step_finish":
        print(f"Step finished: {event.data.get('step_key')}")

    elif event.event_type == "workflow_finish":
        print(f"Workflow completed")

import { defineWorkflow, PolosClient, events } from '@polos/sdk';

const dataPipeline = defineWorkflow<PipelineInput, void, void>(
  { id: 'data-pipeline' },
  async (ctx, input) => {
    const data = await ctx.step.run('extract', () => extractData(input));
    const transformed = await ctx.step.run('transform', () => transformData(data));
    const final = await ctx.step.invokeAndWait('child_pipeline', childPipeline, transformed);
    await ctx.step.run('load', () => loadData(final));
  }
);

// Start workflow
const client = PolosClient.fromEnv();
const handle = await dataPipeline.invoke(
  client,
  { source: 'database', table: 'orders' }
);

// Stream events
for await (const event of events.streamWorkflow(client, handle.rootWorkflowId, handle.id)) {
  if (event.eventType === 'step_start') {
    console.log(`Step started: ${event.data?.step_key}`);
  } else if (event.eventType === 'step_finish') {
    console.log(`Step finished: ${event.data?.step_key}`);
  } else if (event.eventType === 'workflow_finish') {
    console.log('Workflow completed');
  }
}

from polos import events, PolosClient

async def stream_to_websocket(client: PolosClient, root_workflow_id: str, execution_id: str, websocket):
    """Stream workflow events to WebSocket client."""
    async for event in events.stream_workflow(client, root_workflow_id, execution_id):
        # Send event to frontend
        await websocket.send_json({
            "type": event.event_type,
            "data": event.data,
            "timestamp": event.created_at
        })

        # Stop streaming when workflow completes
        if event.event_type == "workflow_finish" and event.data["metadata"]["execution_id"] == execution_id:
            break

import { events, PolosClient } from '@polos/sdk';

async function streamToWebsocket(
  client: PolosClient,
  rootWorkflowId: string,
  executionId: string,
  websocket: WebSocket
) {
  for await (const event of events.streamWorkflow(client, rootWorkflowId, executionId)) {
    // Send event to frontend
    websocket.send(JSON.stringify({
      type: event.eventType,
      data: event.data,
      timestamp: event.createdAt,
    }));

    // Stop streaming when workflow completes
    if (event.eventType === 'workflow_finish' && event.data?.metadata?.execution_id === executionId) {
      break;
    }
  }
}