Developer Guide

Creating nodes

To create your own engine, you must execute the following steps:

Own engine repository

If you don't have an engine to host your new nodes already, create a fork of the Saturn-Engine.

Setup new Engine

git clone <Engine-URL> --recurse-submodules

The engine is the place that hosts all the implementations of the nodes that can be used.

Creating Nodes

All Nodes that can be accessed by Saturn must be subclasses of the Node class.

To create a new node, make a new file inside the nodes/ directory within the saturn_engine/ module. In that file, create a new class that derives from the Node class. Now, all the remaining steps take place within this new class.

Naming and documentation

Make sure, your new class has a descriptive name, since the name displayed within your graphs is derived from that name.

To provide a description about what the node does, simple write a DocString.

nodes/integer_div.py

from node import Node

 

class IntegerDiv(Node):

"""Divides 2 integer numbers. Behaves like the python `//` operator."""

...

Mandetory implementations

To quickly get running, there is only one method you need to implement for your node: the run() method.

In this method, you can basically do what you want. You can also specify any number of input and output parameters (more on that later) and even write additional helper methods you can call from that node.

Annotating parameters

To properly display your node and add type support, you should annotate the input and output parameters. Here is a guide on how to do it:

If you only want to provide type information and nothing more (like a description or constarints), you can simply add normal Python TypeHints. If you want to provide additional type information, use the Annotated type. For the first parameter of Annotated then use a classical TypeHint as you would normally do. For the metadata, you can then add an object of a BaseSchema subclass. There are many different subclasses of BaseSchema that allow for different addition information.

E.g., say, we have a node that takes a string for an input, but we know, that this string has to match a given pattern. Then we can use the StringSchema to specify that pattern and provide further type information.

nodes/integer_div.py

from node import Node

from shared.schema import IntegerSchema

 

class IntegerDiv(Node):

"""Divides 2 integer numbers. Behaves like the python `//` operator."""

 

def run(nom: int, denom: Annotated[int, IntegerSchema(minimum=1)]) -> int:

return nom // denom

Tips for annotations

Certain mechanics are implemented, so you don't have to write as much when annotating your parameters:

Descriptions

If the parameter is already properly typed and you only want to add a description as well, you don't have to construct the proper schema object as well.

Instead you can simply use an AnySchema and only provide the description:

Annotated[int | str, AnySchema(name="BetterName", description="Some extra description")]

 

# is the same as

 

Annotated[int | str, UnionSchema(parameters=[IntegerSchema(), StringSchema()], name="BetterName", description="Some extra description")]

Info

This is because the schema inferred from int | str is more descriptive than the AnySchema, but since it has no name or description set, those will be merge in from the provided AnySchema.

Contradicting Schemas

There is still a real chance of having contradicting schemas. E.g., using BoolSchema here would cause an error.

Using File-References

To use files across engines, you must add the following constructor to your node class:

def __init__(self, data_layer: DataLayer):

self.data_layer = data_layer

Now you can access files in the following way:

from node import Node

from shared.data import DataFileContext, DataLayer

from shared.schema import RefBase

 

class FileRef(RefBase): # (1)

id: str

 

 

class MyNode(Node):

def __init__(self, data_layer: DataLayer):

self.data_layer = data_layer

 

async def run(self, file_ref: FileRef):

file = await self.data_layer.assert_get_file(file_ref.id)

 

async with self.data_layer.context(file) as ctx:

file_path = ctx.path

  1. We add a custom FileRef expanding RefBase, since RefBase doesn't contain a id.

    If you want to use refs from other engines, you might need to copy those into your code-base.

The DataLayer supports other methods as well. E.g., for copying files or creating files from a given content.

Optional Implementations

Sometimes we can provide further type-information during run-time. Such as available references or the like.

To provide such dynamic type information, we can also implement the update_input_schema() and update_output_schema() methods of our node class.

Those to functions are called during the construction of the graph to provide type information in real time.

Note

Note, you do not have to implement those functions, but when your node can provide additional type info based on its input or the environment, it might be helpful to add an implementation

Let's say, we have the following node:

class ReadFile(Node):

"""Reads the contents of the given file inside the `public/` directory"""

 

def run(self, path: str) -> str:

...

During runtime, we could provide the user with some additional constraints about the available files, so they can only enter existing filenames.

For such a case, we could implement the update_input_schema() method:

class ReadFile(Node):

"""Reads the contents of the given file inside the `public/` directory"""

 

def run(self, path: str) -> str:

...

 

def update_input_schema(self) -> ObjectSchema:

f = []

# collect all the available files

for (_, _, files) in os.walk("public"):

f.extend(files)

break

 

# Create an enum variant for each file

files = [

EnumSchemaValue(value=file, label=file) for file in f

]

 

# Return a new input schema updating the schema of some input parameters

return ObjectSchema(

fields: {

"path": EnumSchema(possible_values=files) (3)

}

)

Logging

The Node base class provides a logger member that can be used for logging:

class LoggingNode(Node):

 

def run(self):

self.logger.info("Info log message")