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.
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.
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
-
We add a custom
FileRefexpandingRefBase, sinceRefBasedoesn'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")