This document provides comprehensive patterns and best practices for developing Python nodes in Workflows. It covers various development patterns, error handling, performance considerations, and advanced techniques.

Overview

Developing Python nodes for Workflows requires understanding not just the basic API, but also the patterns and best practices that make nodes robust, efficient, and maintainable.

Basic Development Patterns

Simple Node Pattern

A simple node with no inputs or outputs:

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

Node with Properties

A node that uses properties:

from Mistika.Qt import QColor
 
def init(self):
    self.color = QColor(225, 225, 210)
    self.addProperty("watchers") 
    self.addProperty("deliveries")
    return True
 
def isReady(self):
    res=True
    if self.watchers=="":
        res=self.critical("config:watchers","'Watchers' folder can not be empty") and res
    if self.deliveries=="":
        res=self.critical("config:deliveries","'Deliveries' folder can not be empty") and res
    return res
 
def process(self):
    return True

Node with Connectors

A node that handles input/output connectors:

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.bypassSupported = True
    return True
 
def isReady(self):
    if self.bypassSupported and self.bypassEnabled:
        return True    
    return True
 
def process(self):
    if self.bypassEnabled:
        return True
        
    inputs = self.getConnectorsByType(Cconnector.CONNECTOR_TYPE_INPUT, Cconnector.CONNECTOR_SPECIALTYPE_NORMAL)
    output = self.getFirstConnectorByName("output")
    
    # Process inputs and add to output
    for c in inputs:
        for up in c.getUniversalPaths():
            # Process each universal path
            pass
    
    return True

Error Handling

Proper error handling is crucial for robust nodes:

Critical Errors

Use self.critical() for critical errors that prevent node execution:

def isReady(self):
    if not os.path.exists(self.ffmpegPath()):
        return self.critical("ffmpeg:isReady","ffmpeg binary not found: ".format(fp))
    return True

Graceful Failures

Handle failures gracefully and provide meaningful error messages:

def process(self):
    try:
        # Processing logic
        pass
    except Exception as e:
        self.critical("process", "Error processing: {}".format(str(e)))
        return False
    return True

Progress Tracking

Proper progress tracking improves user experience:

Set Complexity

Set the complexity of the node for accurate progress reporting:

def process(self):
    self.setComplexity(totalNumberOfUPs(self)*100)
    # ... processing ...
    self.progressUpdated(self.complexity())

Update Progress

Update progress at logical intervals:

def process(self):
    currentProgress = 0
    for item in items:
        # Process item
        currentProgress += 1
        self.progressUpdated(currentProgress)

Cancellation Support

Nodes should support cancellation to allow users to stop long-running operations:

def process(self):
    for item in items:
        if self.isCancelled():
            return False
        # Process item

Connector Handling

Input Connector Handling

Process input connectors properly:

def process(self):
    inputs = self.getConnectorsByType(Cconnector.CONNECTOR_TYPE_INPUT, Cconnector.CONNECTOR_SPECIALTYPE_NORMAL)
    for c in inputs:
        for up in c.getUniversalPaths():
            if self.isCancelled():
                return False
            # Process each universal path
            print(up.getFileName(), ": ", up.toString())

Output Connector Handling

Manage output connectors:

def process(self):
    output = self.getFirstConnectorByName("output")
    output.clearUniversalPaths()  # Clear previous outputs
    
    # Add new universal paths
    output.addUniversalPath(new_up)

External Application Integration

Basic External App Integration

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

Advanced External App Integration

def process(self):
    apps = []
    queue = self.createExecutionQueue(threads)
    
    for item in items:
        app = self.createExternalApp()
        app.setParams(binary_path, arguments)
        apps.append(app)
        queue.addJob(app)
    
    while queue.wait(1000):
        if self.isCancelled():
            queue.cancelAll()
            break
    
    # Process results
    for app in apps:
        if app.exitCode():
            self.critical("process", "External app failed")
        app.deleteLater()
    
    queue.deleteLater()

Property Validation

Validate properties in the isReady function:

def isReady(self):
    if not self.url:
        return self.critical("process", "URL is required")
    
    if self.threads < 1:
        return self.critical("process", "Threads must be at least 1")
    
    return True

Bypass Functionality

Implement bypass functionality when appropriate:

def init(self):
    self.bypassSupported = True
    return True
 
def isReady(self):
    if self.bypassSupported and self.bypassEnabled:
        return True
    # Normal readiness checks
    return True
 
def process(self):
    if self.bypassEnabled:
        return True
    # Normal processing logic
    return True

Logging and Debugging

Use appropriate logging for debugging:

def process(self):
    self.info("process", "Starting processing")
    self.info("process", "Processing {} items".format(len(items)))
    
    for i, item in enumerate(items):
        self.info("process", "Processing item {}".format(i))
        # Process item
    
    self.info("process", "Processing completed")

Performance Considerations

Memory Management

Properly manage memory by cleaning up resources:

def process(self):
    apps = []
    # ... processing ...
    for app in apps:
        app.deleteLater()
    queue.deleteLater()

Thread Management

Use appropriate thread counts:

def process(self):
    threads = max(1, int(self.threads))  # Ensure at least 1 thread
    queue = self.createExecutionQueue(threads)

Testing and Debugging

Unit Testing

While direct unit testing of nodes is difficult, you can test individual functions:

# Test property validation
def test_isReady_with_empty_url():
    node = MyNode()
    node.url = ""
    result = node.isReady()
    assert result == False

Workflow Testing

Test nodes in actual workflows:

Add node to workflow
Connect inputs/outputs
Run workflow
Check logs for errors
Verify outputs

Advanced Development Patterns

Conditional Processing

def process(self):
    if self.condition1:
        # Process with condition 1
        pass
    elif self.condition2:
        # Process with condition 2
        pass
    else:
        # Default processing
        pass

Batch Processing

def process(self):
    batch_size = 10
    for i in range(0, len(items), batch_size):
        batch = items[i:i + batch_size]
        # Process batch
        self.progressUpdated(i + len(batch))

Configuration Loading

def init(self):
    self.addProperty("configFile")
    self.addProperty("configData")
    return True
 
def isReady(self):
    if self.configFile:
        # Load configuration
        try:
            self.configData = load_config(self.configFile)
        except Exception as e:
            return self.critical("config", "Failed to load config: {}".format(str(e)))
    return True

Node Naming Conventions

Follow consistent naming conventions:

Use descriptive names that indicate functionality
Use PascalCase for node names
Keep names concise but meaningful

Examples:

FFmpeg
GoogleDriveIn
MetadataToCSV
Change8KP2metaData

Schema Integration

The XSD schema file should match the node name in lowercase:

<!-- For node named FFmpeg.py -->
<!-- Schema file should be ffmpeg.xsd -->
<xsd:element name="FFmpeg" sgo:categoryType="categoryCollapsible" 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="url" type="nameConventionFilePathTextEditor" sgo:visibility="0" sgo:sortOrder="4"/>
            <xsd:element name="dstExtension" type="xsd:string" sgo:visibility="0" sgo:sortOrder="5"/>
            <xsd:element name="threads" type="integerPulldown" sgo:integerPulldownValues="1|2|4" sgo:visibility="0" sgo:sortOrder="6"/>
        </xsd:sequence>
    </xsd:complexType>
</xsd:element>

Common Mistakes to Avoid

Not implementing all required functions (init, isReady, process)
Forgetting to set self.bypassSupported = True when implementing bypass
Not handling self.isCancelled() properly
Not setting self.setComplexity() for progress tracking
Not properly cleaning up external applications and queues
Not validating properties in isReady
Not using appropriate error messages with self.critical()
Not handling connector types correctly
Not following naming conventions
Not testing in actual workflows

Summary

Creating robust Python nodes for Workflows requires:

Understanding the basic API and required functions
Implementing proper error handling
Supporting cancellation and progress tracking
Following best practices for performance and resource management
Testing thoroughly in actual workflows
Maintaining consistency with existing nodes

By following these patterns and best practices, you'll create nodes that are reliable, efficient, and easy to maintain.