Creating Python Nodes for Workflows

This document provides comprehensive guidance on creating Python nodes for the Workflows system. Python nodes are the core building blocks that define custom functionality within workflows.

Overview

Python nodes in Workflows are implemented as Python files that define specific functions to handle initialization, readiness checking, and processing logic. Each node file must contain at least three required functions: init, isReady, and process.

Basic Node Structure

A minimal Python node looks like this:

def init(self):
    return True
    
def isReady(self):
    return True
    
def process(self):
    # Your processing logic here
    return True

Required Functions

init()

The init function is called when the node is initialized. This is where you set up the node's properties, connectors, and other configurations.

Key methods available in init

• self.addConnector(name, type, mode) - Add input/output connectors
• self.addProperty(name, defaultValue) - Add a property with optional default value. If the property name starts with an underscore (_), it will be created as internal and will not be stored in the .mwf workflow file.
• self.setAcceptConnectors(accept, pattern) - Set which connectors are accepted
• self.setDropToProperty(property_name) - Set which property receives dropped files
• self.setDropSupportedTypes(types) - Set supported drop types (1=Directories, 2=Files, 3=both)
• self.bypassSupported = True - Enable bypass functionality
• self.color = QColor(r, g, b) - Set node color

isReady()

The isReady function checks if the node is ready to process. This is called before process

Return values:

• True - Node is ready to process
• False - Node is not ready (will be marked as error)
• self.critical("message", "error description") - Return critical error

process()

The process function contains the main logic of the node. This is where the actual work happens.

Key methods available in process

• self.getConnectorsByType(type, specialType) - Get connectors by type
• self.getFirstConnectorByName(name) - Get first connector by name
• self.evaluate(property_name) - Evaluate a property value
• self.isCancelled() - Check if node execution was cancelled
• self.progressUpdated(value) - Update progress
• self.setComplexity(value) - Set node complexity for progress tracking
• self.info("category", "message") - Log information
• self.critical("category", "error message") - Log critical error
• self.addFailedUP(up, message) - Add failed universal path

Handling Persistent Errors

In some cases, you may need to report errors that should persist beyond the normal validation process. This is particularly useful when errors occur outside of the normal validation flow (e.g., in menuAction or processDroppedFile).

The functions self.addPermanentCritical() and self.addPermanentError() are designed for this purpose:

• self.addPermanentCritical(id, description, group) - Add a permanent critical error
• self.addPermanentError(id, description, group) - Add a permanent error

These functions work similarly to their regular counterparts but:

1. The errors persist beyond the normal validation process
2. They are not automatically cleared when checking isReady
3. They remain active until explicitly cleared using:
   • self.clearPermanentErrors() - Clear all permanent errors
   • self.deletePermanentErrorById(id) - Delete error by ID
   • self.deletePermanentErrorsByGroup(group) - Delete errors by group

Important: For errors in init, isReady, and process functions, use the regular critical and error methods instead. Permanent errors should only be used in functions that are not part of the core validation/execution process.

Example usage in menuAction:

def menuAction(self, actionName):
    if actionName == "Authenticate":
        # Perform authentication
        refreshToken = goo.getRefreshToken(self.clientId.strip(), self.clientSecret.strip(), 60)
        
        if refreshToken == "TimeOut":
            return self.addPermanentCritical("GoogleMail:process:authCanceled", "Authentication timeout")
        elif refreshToken == None:
            return self.addPermanentCritical("GoogleMail:process:authCanceled", "Authentication was rejected")
        else:
            # Clear any previous authentication errors
            self.deletePermanentErrorById("GoogleMail:process:authCanceled")
            self.refreshToken = refreshToken

Example usage in processDroppedFile:

def processDroppedFile(self, filePath):
    # Validate file type
    if not filePath.endswith(('.mov', '.mp4')):
        # Report persistent error that won't be cleared automatically
        self.addPermanentError("file:invalid_type", "Only MOV and MP4 files are supported")
        return False
    # Clear error if file is valid
    self.deletePermanentErrorById("file:invalid_type")
    # Process the file
    self.url = filePath
    return True

Node Properties

Properties are defined using self.addProperty(name, defaultValue) and can be accessed directly as self.property_name.

Connectors

Connectors are used to connect nodes in a workflow. Types:

• Cconnector.CONNECTOR_TYPE_INPUT - Input connector
• Cconnector.CONNECTOR_TYPE_OUTPUT - Output connector
• Cconnector.MODE_OPTIONAL - Optional connector
• Cconnector.MODE_REQUIRED - Mandatory connector

Example Node Patterns

Simple Node (No Inputs/Outputs)

def init(self):
    return True
    
def isReady(self):
    return True
    
def process(self):
    self.info("HelloWorld","Hello World")
    return True

Node with Connectors and Properties

from Mistika.Qt import QColor
from Mistika.classes import Cconnector
 
def init(self):
    self.color = QColor(92, 184, 92)
    self.addConnector("input", Cconnector.CONNECTOR_TYPE_INPUT, Cconnector.MODE_OPTIONAL)    
    self.addConnector("output", Cconnector.CONNECTOR_TYPE_OUTPUT, Cconnector.MODE_OPTIONAL)    
    self.setAcceptConnectors(True, "input_%") 
    self.addProperty("url")    
    self.addProperty("threads", 1)
    self.bypassSupported = True
    return True
 
def isReady(self):
    if self.bypassSupported and self.bypassEnabled:
        return True    
    # Add readiness checks here
    return True
 
def process(self):
    # Process logic here
    return True

XSD Schema Integration

Each node has an associated XSD schema file that defines the properties shown in the node editor. Note: The properties should be created in the @init function. The XSD defines only whitch properties to show and the format to use.

The schema file name matches the node name in lowercase.

Example schema structure:

<xsd:schema xmlns:sgo="http://www.sgo.es/xsd/2026/sgo.xsd"
            xmlns:xsd="http://www.w3.org/2001/XMLSchema">
<xsd:element sgo:categoryType="categoryCollapsible" name="NodeName" sgo:visibility="0" sgo:sortOrder="0">
    <xsd:complexType>
        <xsd:sequence>
            <xsd:element name="objectName" type="xsd:string" sgo:visibility="0" sgo:sortOrder="0"/>
            <xsd:element name="color" type="color" sgo:visibility="0" sgo:sortOrder="1"/>
            <xsd:element name="bypassEnabled" type="xsd:boolean" sgo:visibility="0" sgo:sortOrder="2"/>
            <xsd:element name="nodeLocked" type="xsd:boolean" sgo:visibility="0" sgo:sortOrder="3"/>
            <xsd:element name="property1" type="xsd:string" sgo:visibility="0" sgo:sortOrder="4"/>
            <xsd:element name="property2" type="integerPulldown" sgo:integerPulldownValues="1|2|4" sgo:visibility="0" sgo:sortOrder="5"/>
        </xsd:sequence>
    </xsd:complexType>
</xsd:element>
</xsd:schema>

Best Practices

1. Always implement all three required functions (init, isReady, process)
2. Use proper error handling with self.critical() for critical errors
3. Implement progress tracking for long-running operations
4. Check self.isCancelled() regularly to support cancellation
5. Set appropriate node colors for visual identification
6. Use meaningful property names and descriptions
7. Handle both input and output connectors properly
8. Implement bypass functionality when appropriate
9. Use self.setComplexity() for accurate progress reporting
10. Follow the naming conventions used in existing nodes
11. Use self.addPermanentCritical() and self.addPermanentError() for persistent errors that should not be cleared automatically

Common Patterns

Progress Tracking

self.setComplexity(totalNumberOfUPs(self)*100)
// ... processing ...
self.progressUpdated(self.complexity())

Connector Handling

inputs = self.getConnectorsByType(Cconnector.CONNECTOR_TYPE_INPUT, Cconnector.CONNECTOR_SPECIALTYPE_NORMAL)
for c in inputs:
    for up in c.getUniversalPaths():
        // Process each universal path

External Application Integration

app = self.createExternalApp()
app.setParams(binary_path, arguments)
queue = self.createExecutionQueue(threads)
queue.addJob(app)
queue.wait()

Development Workflow

1. Create a new Python file in /workspace/project/src/data/resources.workflows.nodes.library/shared/workflowsLibrary/
2. Implement the required functions (init, isReady, process)
3. Create corresponding XSD schema file in /workspace/project/src/data/resources.workflows.nodes.library/shared/config/schemas/
4. Test the node in the workflow editor
5. Debug and refine as needed

Node Testing

Nodes can be tested by:

1. Adding them to a workflow in the editor
2. Connecting appropriate inputs/outputs
3. Running the workflow to observe behavior
4. Checking logs for error messages
5. Verifying that properties are correctly set and accessible

Troubleshooting

Common issues:

• Missing required functions (init, isReady, process)
• Incorrect connector setup
• Property access errors
• Progress tracking issues
• External application path problems
• Schema validation errors
• Incorrect use of error reporting functions

When using error reporting functions:

• Use regular self.critical(), self.error(), self.warning(), self.info() for errors that should be cleared automatically during validation
• Use self.addPermanentCritical() and self.addPermanentError() for errors that should persist beyond the normal validation process (e.g., in menuAction or processDroppedFile)

Always check the workflow logs for error messages when debugging node issues.