Dflow

API

A Dflow represents a program as an executable graph.

A graph can contain nodes and links. Nodes are executed, sorted by their connections.

Constructor

Dflow constructor requires a list of node definitions which is an Array<DflowNode>.

Methods

dflow.node(kind)

Create a new node. Returns node id.

dflow.link(source, target)

Create a new link and connect two nodes. Returns link id.

The source and target arguments can be either:

An array [nodeId, position]: The source node id and the output position or the target node id and the input position
A string: The source or target node id. The position defaults then to 0 i.e. the first one

dflow.data(value)

Create a new data node. Returns node id.

If value is not a valid DflowData, it will be set to undefined.

dflow.run()

Execute all nodes, sorted by their connections.

Notice that the run method is async and returns a Promise<void>. In general you should call it with await dflow.run() but if no node in the graph is async, then you can just call it with dflow.run().

dflow.delete(itemId)

Delete node or link with given id.

The itemId argument can be either a node id or a link id. If it is a node id, then all links connected to that node will be deleted as well as the outputs related to that node.

Getters

dflow.graph

A graph contains nodes and links.

A DflowGraph can be serialized into JSON so it can be saved and loaded by a Dflow instance.

It has the following attributes:

node: Record<string, string>: Key is node id, value is node kind
link: Record<string, DflowLink>: Key is link id
data: Record<string, DflowData>: Data nodes: key is node id, value is node kind

dflow.error

Get error messages from last run, indexed by node id.

dflow.out

Get output data of last run, indexed by node id.

Static methods

Dflow.input(typing, options?)

Helper to define inputs.

It is supposed to be used inside a node definition. For example, define an input string named message.


const Print: DflowNode = {
  kind: "print",
  inputs: [Dflow.input("string", { name: "message" })],
  run: (message: string) => {
    console.log(message);
  }
};

Input with number type.


Dflow.input("number")

Input that accepts both number and string type.


Dflow.input(["number", "string"])

Input with type array and name.


Dflow.input("array", { name: "list" })

Input that accepts any type.


Dflow.input()

Input with any type and named "foo".


Dflow.input([], { name: "foo" })

Optional number input.


Dflow.input("number", { optional: true })

Dflow.output(typing, options?)

Helper to define outputs.

The signature is similar to Dflow.input() except for the optional parameter, which is not needed for outputs.

It is supposed to be used inside a node definition. For example, define a number output named π (PI).


const MathPI: DflowNode = {
  kind: "mathPI",
  outputs: [Dflow.output("number", { name: "π" })],
  run: () => Math.PI
};

Types

DflowData

Includes JSON data types and undefined

The DflowData can be one of the following:

undefined
null
boolean
number
string
DflowArray
DflowObject

Where DflowArray is just an array of DflowData and DflowObject is an object with string keys and DflowData values.

DflowDataType

Dflow data types represent values that can be serialized as JSON.

The DflowDataType is a literal type; it can be one of the following:

"null"
"boolean"
"number"
"string"
"array"
"object"

Both DflowInput and DflowOutput have a types: DflowDataType[] attribute which is used to check if they can be connected.

A special case is when types is an empty array. In this case, the input or output can accept any data type.

DflowInput

Defines a node input.

You can use Dflow.input() helper to create them.

A DflowInput has the following attributes:

name?: string: Ignored by Dflow, but could be used by UI.
types: DflowDataType[]: An input can be connected to an output only if the data types match.
optional?: boolean: An input is required by default. If it is not connected or the data passed is not valid according to its types, then its node will not be executed. If an input is optional the data checks are skipped.

DflowOutput

Defines a node output.

You can use Dflow.output() helper to create them.

A DflowOutput has the following attributes:

name?: string: Ignored by Dflow, but could be used by UI.
types: DflowDataType[]: An output can be connected to an input only if the data types match.

DflowLink

Connects two nodes in the graph.

A DflowLink is a list with four elements:

sourceNodeId: string: The id of output node
sourcePosition: number: it defaults to 0 if not provided in dflow.link()
targetNodeId: string: The id of input node
targetPosition: number: it defaults to 0 if not provided in dflow.link()

DflowNode

Defines a block of code: it can have inputs and outputs.

Dflow does not provide a set of pre-defined nodes. You must implement your own nodes.

For example, a node "addition" could be implemented using BigInt or some arbitrary-precision library, according to your needs... but it is as easy as creating a function. Basically, a node is an object with a run method and a few metadata properties.

See the examples/nodes folder: it contains few node definitions used by dflow tests.

A DflowNode has the following attributes:

kind: string
inputs?: DflowInput[]
outputs?: DflowOutput[]
run(inputs): outputs

When you define a set of nodes, you may want to start the file with something like this:


import { Dflow, type DflowNode } from "dflow";

const { input, output } = Dflow;

So you have the DflowNode type that can help you define nodes and the input and output helpers to define inputs and outputs.

This is a node that implements Array.prototype.join() so its first input is an array and it has an optional second input for the separator, which defaults to a comma.


const ArrayJoin: DflowNode = {
  kind: "arrayJoin",
  inputs: [
    input("array"),
    input("string", { name: "separator", optional: true })
  ],
  outputs: [output("string")],
  run(array: DflowArray, separator: string | undefined) {
    return array.join(separator);
  }
};

First of all, the name attribute of inputs and outputs is optional and ignored by Dflow. It could be used by a UI to show meaningful names. So there is no need to match the input names with the run method arguments.

Notice that the run method arguments are typed according to the inputs definition. You can import DflowArray type from dflow package to type the first argument. The second argument is either a string or undefined, because the input is marked as an optional string.

It is recommended to just use the types defined by DflowData. This is the thing: the node above will run only if the first input is connected to an output that produces an array and the second input is either not connected or connected to an output that produces a string.

It is also recommended to not type the return value, just let TypeScript infer it.

Here is an example of nodes generated from the Math global.


import { Dflow } from "dflow";

// Generate Dflow nodes for all Math properties and functions.

const mathNodes = Object.getOwnPropertyNames(Math).map((key) => {
  // @ts-expect-error: expression of type 'string' can't be used to index type 'Math'
  const item = Math[key];

  const kind = `Math.${key}`;
  const outputs = [Dflow.output("number")];

  // If the item is a number, create a node that outputs that number.
  if (typeof item === "number") {
    return {
      kind,
      outputs,
      run: () => item
    };
  }

  // If the item is a function, wrap in in the run method.
  if (typeof item === "function") {
    return {
      kind,
      // Get the number of inputs from the function's length property.
      inputs: Array(item.length).fill(Dflow.input("number")),
      outputs,
      run: (...args: number[]) => {
        return item(...args);
      }
    };
  }
  // Not needed, just to make TS happy.
  throw new Error(`Unsupported Math property: ${key}`);
});

// Create a Dflow instance with the generated node definitions.
const dflow = new Dflow(mathNodes);

// Compute Math.trunc(Math.E)
const nodeId1 = dflow.node("Math.E");
const nodeId2 = dflow.node("Math.trunc");
dflow.link(nodeId1, nodeId2);

dflow.run();

console.log(dflow.out);
// { n0: [ 2.718281828459045 ], n1: [ 2 ] }

Miscellanea

dflow.context

Dflow context is bound to every node at runtime, hence it is accessible via this inside node run.


// Create a host with an API context.
import { Dflow, type DflowNode } from "dflow";

// Add an API client to the context.
// A Dflow context is a Record that will be bound to nodes at runtime.

type Context = {
  apiClient: ApiClient;
};

// Of course this is a dummy API client.
class ApiClient {
  apiKey: string;

  constructor(apiKey: ApiClient["apiKey"]) {
    this.apiKey = apiKey;
  }

  async fetchSomeData(
    payload: string
  ): Promise<{ status: string; payload: string }> {
    return await Promise.resolve({ status: "SUCCESS", payload });
  }
}

// This nodes uses the apiClient from the context...
const CustomNode: DflowNode = {
  kind: "Custom",
  inputs: [Dflow.input("string")],
  outputs: [Dflow.output("object")],
  // ... notice that we specify the type of `this` via the
  //
  //     this: Context
  //
  // argument on the run method.
  async run(this: Context, data: string) {
    const result = await this.apiClient.fetchSomeData(data);
    return result;
  }
};

// Create a Dflow instance and add the context.
const dflow = new Dflow([CustomNode]);
dflow.context.apiClient = new ApiClient("s3cret");

const nodeId = dflow.node(CustomNode.kind);
const dataId = dflow.data("foo");
dflow.link(dataId, nodeId);

await dflow.run();

const result = dflow.out[nodeId][0] as { status: string; payload: string };

if (!result || result.status !== "SUCCESS" || result.payload !== "foo")
  console.error("Unexpected result:", result);
else console.info(result.status);

dflow.ERR

Optional error logger.

By default, Dflow does not log errors. For example to send errors to STDERR, you can do:


dflow.ERR = console.error;

The logger must be a function that accepts a single argument of any type. Usually it will be an Error object but it could be anything that can be thrown by a node, for example a string.