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
- 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.
- 2Write two sample orders and inspect the materialized source for its exact table name.
- 3Create a flow, add one transform, then read its source state before saving the complete handler.
- 4Run the draft and wait for that workflow to complete. Check the exact run output equals 350 before publishing.
- 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
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 waitForRun(id: string) {
const deadline = Date.now() + 5 * 60_000;
while (Date.now() < deadline) {
const { items } = await client.operations.listRuns(id);
const run = items.find((item) => item.workflowId === started.workflowId);
if (run?.status === 'failed') throw new Error(run.error ?? 'Run failed');
if (run?.status === 'completed') return run;
await new Promise((resolve) => setTimeout(resolve, 2000));
}
throw new Error('Timed out waiting for ' + id);
}
const [run] = await Promise.all([waitForRun(operationId), waitForRun(flow.operation.id)]);
if (!run.outputDocId) throw new Error('Run output is missing');
const preview = await client.operations.getRunOutput(operationId, run.id);
console.table(preview.tables.revenue);
const rows = preview.tables.revenue as Array<{ total: unknown }> | undefined;
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 the latest graph revision.
flow = await client.dataFlows.get(flow.operation.id);
await client.dataFlows.command(flow.operation.id, {
command: { type: 'publish', expectedRevision: flow.revision },
});
console.log('Published flow:', flow.operation.id, 'Output source:', operationId);