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")

    On this page