PermuteDocs
Go to Permute

Guide

Build and publish a data flow

Turn two sample orders into a published revenue total of 350. This complete SDK example creates the input, saves a transform, runs the flow, checks its output, and publishes it (no built-in coding agent required).

Workflow

  1. 1Install @permute/sdk with bun add @permute/sdk. Set PERMUTE_API_KEY and PERMUTE_WORKSPACE_ID for a workspace where the key can read and write data and operations. Save the example as data-flow.ts and run bun data-flow.ts.
  2. 2Write two sample orders and inspect the materialized source for its exact table name.
  3. 3Create a flow, add one transform, then read its source state before saving the complete handler.
  4. 4Run the draft and wait for that workflow to complete. Check the exact run output equals 350 before publishing.
  5. 5For later edits, repeat read → save → run → inspect → publish. On a 409 conflict, read again and reconcile your changes before retrying. Each new run has its own output; published operations are available through sources and queries.

Complete example

typescript
import { PermuteClient } from '@permute/sdk';

const client = new PermuteClient({
  apiKey: process.env.PERMUTE_API_KEY!,
}).withWorkspace(process.env.PERMUTE_WORKSPACE_ID!);

// 1. Create sample input and wait for materialization.
const dataset = await client.datasets.create({
  type: 'custom', key: 'data-flow-guide', name: 'Sample orders', context: [],
});
await dataset.replace({ orders: [{ amount: 100 }, { amount: 250 }] });
const input = await client.sources.get(dataset.id);
const table = input.tables[0];
if (!table) throw new Error('Sample table is missing');
const sql = 'SELECT SUM(amount) AS total FROM "' + table.name.replaceAll('"', '""') + '"';

// 2. Add a transform to a new flow.
let flow = await client.dataFlows.create({
  name: 'Revenue total', description: 'Sum sample orders', dataSourceIds: [dataset.id],
});
await client.dataFlows.command(flow.operation.id, {
  command: {
    type: 'addTransform', expectedRevision: flow.revision,
    input: { name: 'Total', instruction: 'Sum order amounts', inputIds: [dataset.id] },
  },
});
flow = await client.dataFlows.get(flow.operation.id);
const operationId = flow.operationIds[0];
if (!operationId) throw new Error('Transform is missing');

// 3. Save code using the state we just read.
const current = await client.operations.getSource(operationId);
await client.operations.updateSource(operationId, {
  source: 'module.exports = { handler: async (event) => ({ revenue: await event.query(' +
    JSON.stringify(sql) + ') }) };',
  expectedRevision: current.revision,
  expectedDataFlowRevision: current.dataFlowRevision,
  expectedUpdatedAt: current.updatedAt,
});

// 4. Run the whole draft and wait for this workflow, not an older run.
const started = await client.operations.run(flow.operation.id, { version: 'draft' });
async function waitForWorkflow() {
  const deadline = Date.now() + 5 * 60_000;
  while (Date.now() < deadline) {
    const workflow = await client.operations.listRuns({ workflowId: started.workflowId });
    if (workflow.status === 'failed') throw new Error(workflow.items.find((run) => run.error)?.error ?? 'Workflow failed');
    if (workflow.status === 'completed') return workflow;
    await new Promise((resolve) => setTimeout(resolve, 1500));
  }
  throw new Error('Timed out waiting for workflow');
}
const workflow = await waitForWorkflow();
const run = workflow.items.find((run) => run.operationId === operationId);
if (!run?.outputDocId) throw new Error('Run output is missing');
const preview = await client.operations.getRunOutput(operationId, run.id);
const revenueBlockId = Object.entries(preview.names).find(([, name]) => name.toLowerCase() === 'revenue')?.[0];
const rows = revenueBlockId ? preview.tables[revenueBlockId] as Array<{ total: unknown }> : undefined;
console.table(rows);
if (rows?.length !== 1 || Number(rows[0].total) !== 350) {
  throw new Error('Unexpected total; inspect the source and fix the handler before publishing');
}

// 5. Publish only after validating the output, using its current version token.
const ready = await client.revisions.list(flow.operation.id);
await client.revisions.publish(flow.operation.id, { expectedVersion: ready.version });
console.log('Published flow:', flow.operation.id, 'Output source:', operationId);