Protocols¶
Type protocols and metadata utilities for node registration.
NodeFactory(**initial_values)
¶
Bases: Protocol
Protocol defining the interface for node type definitions.
DEPRECATION WARNING: The 'parameters' attribute is deprecated. Instead, define nodes using the grid-based architecture: - grid_layout: Dict defining NodeGrid → GridCell → Components structure - Use HandleComponent (BaseHandle, ButtonHandle, LabeledHandle) for inputs/outputs - Use FieldComponent (TextField, NumberField, etc.) for configuration fields
Any class implementing this protocol can be registered with NodeFlowWidget to create custom node types in the visual editor.
Required Attributes
label (str): Display name for the node shown in UI parameters (Type[BaseModel]): DEPRECATED - Use grid_layout instead
Optional Attributes
icon (str): Unicode emoji or icon identifier (default: "") category (str): Category for grouping nodes in UI (default: "general") description (str): Help text shown to users (default: "") inputs (List[Dict[str, str]]): DEPRECATED - Define in grid_layout instead outputs (List[Dict[str, str]]): DEPRECATED - Define in grid_layout instead grid_layout (Dict[str, Any]): Grid layout configuration (RECOMMENDED)
Required Methods
init: Initialize node instance with optional initial values get_values: Get current configuration values as a dictionary set_values: Update configuration values from a dictionary
Optional Methods
validate: Validate current configuration, returns True if valid execute: Execute node logic (for future execution engine)
Example (deprecated approach): >>> from pydantic import BaseModel >>> >>> class ProcessingConfig(BaseModel): ... threshold: float = 0.5 ... mode: str = "auto" >>> >>> class ImageProcessor: ... label = "Image Processor" ... parameters = ProcessingConfig ... icon = "🖼️"
Example (recommended grid-based approach): >>> from pynodewidget.grid_layouts import create_three_column_grid >>> from pynodewidget.models import ButtonHandle, TextField >>> >>> class ImageProcessor: ... label = "Image Processor" ... icon = "🖼️" ... grid_layout = create_three_column_grid( ... left_components=[ButtonHandle(id="in", label="Input", handle_type="input")], ... center_components=[TextField(id="mode", label="Mode", value="auto")], ... right_components=[ButtonHandle(id="out", label="Output", handle_type="output")] ... )
Initialize node instance with optional initial values.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
**initial_values
|
Any
|
Initial configuration values for the node parameters |
{}
|
Source code in pynodewidget/protocols.py
get_values()
¶
Get current configuration values.
Returns:
| Type | Description |
|---|---|
Dict[str, Any]
|
Dictionary containing all current parameter values |
set_values(values)
¶
Update configuration values.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
values
|
Dict[str, Any]
|
Dictionary of parameter values to update |
required |
validate()
¶
Validate current configuration.
Returns:
| Type | Description |
|---|---|
bool
|
True if configuration is valid, False otherwise |
execute(inputs)
¶
Execute node logic with provided inputs.
This method is optional and used by the execution engine.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
inputs
|
Dict[str, Any]
|
Dictionary of input values from connected nodes |
required |
Returns:
| Type | Description |
|---|---|
Dict[str, Any]
|
Dictionary of output values to pass to connected nodes |
Source code in pynodewidget/protocols.py
HandleSpec
¶
Bases: BaseModel
Specification for a node handle (input or output connection point).
Attributes:
| Name | Type | Description |
|---|---|---|
id |
str
|
Unique identifier for the handle within the node |
label |
str
|
Display name shown in the UI |
handle_type |
Literal['base', 'button', 'labeled']
|
Type of handle component to render ("base", "button", or "labeled") |
NodeMetadata(type_name, label, parameters_schema, icon='', category='general', description='', grid_layout=None)
¶
Metadata extracted from a NodeFactory class for serialization.
This class is used internally to extract and serialize node metadata for transmission to the JavaScript layer.
Initialize node metadata.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
type_name
|
str
|
Unique identifier for the node type |
required |
label
|
str
|
Display name for the node |
required |
parameters_schema
|
Dict[str, Any]
|
JSON Schema for node parameters |
required |
icon
|
str
|
Unicode emoji or icon identifier |
''
|
category
|
str
|
Category for grouping nodes |
'general'
|
description
|
str
|
Help text |
''
|
grid_layout
|
Optional[Dict[str, Any]]
|
Grid layout configuration |
None
|
Source code in pynodewidget/protocols.py
to_dict()
¶
Convert metadata to dictionary for JSON serialization.
Returns:
| Type | Description |
|---|---|
Dict[str, Any]
|
Dictionary representation of node metadata |
Source code in pynodewidget/protocols.py
from_node_class(node_class, type_name=None)
classmethod
¶
Extract metadata from a NodeFactory class.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
node_class
|
Type[NodeFactory]
|
Class implementing NodeFactory protocol |
required |
type_name
|
Optional[str]
|
Optional custom type name (defaults to class name) |
None
|
Returns:
| Type | Description |
|---|---|
NodeMetadata
|
NodeMetadata instance |
Raises:
| Type | Description |
|---|---|
AttributeError
|
If required attributes are missing |
TypeError
|
If parameters is not a Pydantic BaseModel subclass |
Source code in pynodewidget/protocols.py
Overview¶
PyNodeWidget uses Protocols to define interfaces for node classes without requiring inheritance. This provides flexibility for creating custom nodes while maintaining type safety.
NodeFactory Protocol¶
Defines the interface that node classes must implement to be registered with the widget.
Purpose¶
The NodeFactory protocol specifies:
- Required attributes: Minimum properties every node must have
- Optional attributes: Additional customization options
- Required methods: Functions nodes must implement
- Optional methods: Extra functionality nodes can provide
Required Attributes¶
Every node class must define:
from typing import Protocol, ClassVar, Type
from pydantic import BaseModel
class NodeFactory(Protocol):
# Required
type: ClassVar[str] # Unique identifier
category: ClassVar[str] # Organization category
label: ClassVar[str] # Display name
FieldsModel: Type[BaseModel] # Pydantic model for parameters
Optional Attributes¶
Additional customization:
class NodeFactory(Protocol):
# Optional
description: ClassVar[str] # Tooltip description
icon: ClassVar[str] # Lucide icon name or emoji
color: ClassVar[str] # Tailwind color class
# Connection points
inputs: ClassVar[list[str]] # Input handle names
outputs: ClassVar[list[str]] # Output handle names
# Advanced handle configuration
input_specs: ClassVar[list[HandleSpec]]
output_specs: ClassVar[list[HandleSpec]]
# Styling
use_custom_header: ClassVar[bool]
use_custom_footer: ClassVar[bool]
header_class: ClassVar[str]
footer_class: ClassVar[str]
body_class: ClassVar[str]
# Validation
shadow_on_error: ClassVar[str]
errors_at: ClassVar[str]
Required Methods¶
Nodes must implement:
@classmethod
def get_default_values(cls) -> dict:
"""Return default field values."""
return cls.FieldsModel().model_dump()
Optional Methods¶
Additional functionality:
@classmethod
def process(cls, inputs: dict, field_values: dict) -> dict:
"""Process node computation."""
pass
@classmethod
def validate_inputs(cls, inputs: dict) -> bool:
"""Validate input data."""
pass
@classmethod
def render_custom_header(cls, field_values: dict) -> str:
"""Generate custom header content."""
pass
@classmethod
def render_custom_footer(cls, field_values: dict) -> str:
"""Generate custom footer content."""
pass
HandleSpec¶
Pydantic model for defining connection point specifications.
Fields¶
class HandleSpec(BaseModel):
name: str # Handle identifier
label: str # Display name
type: Literal["base", "button", "labeled"] = "base" # Visual style
position: Literal["left", "right", "top", "bottom"] = "left" # Location
color: str = "gray" # Tailwind color
icon: str | None = None # Lucide icon name
description: str | None = None # Tooltip
Usage¶
from pynodeflow.protocols import HandleSpec
input_specs = [
HandleSpec(
name="data",
label="Data Input",
type="labeled",
position="left",
color="blue",
icon="database",
description="Connect data source"
),
HandleSpec(
name="trigger",
label="Trigger",
type="button",
position="top",
color="green",
icon="play"
)
]
Handle Types¶
base: Standard connection point
button: Prominent button-style handle
labeled: Handle with visible label
NodeMetadata¶
Utility class for extracting metadata from node classes.
Purpose¶
NodeMetadata inspects a node class and:
- Extracts all attributes defined by NodeFactory protocol
- Generates JSON schema from FieldsModel
- Provides serializable metadata for JavaScript
- Handles missing optional attributes gracefully
Usage¶
from pynodeflow.protocols import NodeMetadata
class MyNode:
type = "my-node"
category = "data"
label = "My Node"
class FieldsModel(BaseModel):
value: int = 0
# Extract metadata
metadata = NodeMetadata.from_class(MyNode)
# Serialize to dict
data = metadata.model_dump()
# {
# "type": "my-node",
# "category": "data",
# "label": "My Node",
# "fields_schema": {...},
# "default_values": {"value": 0},
# ...
# }
Fields¶
All attributes from NodeFactory protocol, plus:
class NodeMetadata(BaseModel):
# Core
type: str
category: str
label: str
fields_schema: dict # JSON Schema from FieldsModel
default_values: dict # From get_default_values()
# Optional
description: str | None
icon: str | None
color: str | None
# Handles
inputs: list[str]
outputs: list[str]
input_specs: list[HandleSpec]
output_specs: list[HandleSpec]
# Styling
use_custom_header: bool
use_custom_footer: bool
header_class: str | None
footer_class: str | None
body_class: str | None
# Validation
shadow_on_error: str | None
errors_at: str | None
From Class¶
Extracts all available attributes and generates schema.
To Dict¶
Returns serializable dictionary suitable for JSON transmission to JavaScript.
Creating Nodes with Protocols¶
Minimal Node¶
from pydantic import BaseModel
class MinimalNode:
"""Implements required attributes only."""
type = "minimal"
category = "basic"
label = "Minimal Node"
class FieldsModel(BaseModel):
value: str = ""
@classmethod
def get_default_values(cls):
return cls.FieldsModel().model_dump()
Full-Featured Node¶
from pydantic import BaseModel, Field
from pynodeflow.protocols import HandleSpec
class FullNode:
"""Implements all protocol attributes."""
# Required
type = "full-node"
category = "processing"
label = "Full Node"
# Optional metadata
description = "A fully-featured node example"
icon = "box"
color = "blue"
# Simple handles
inputs = ["input"]
outputs = ["output", "error"]
# Advanced handle specs
input_specs = [
HandleSpec(
name="input",
label="Data Input",
type="labeled",
color="blue",
icon="database"
)
]
output_specs = [
HandleSpec(
name="output",
label="Success",
type="labeled",
color="green",
icon="check"
),
HandleSpec(
name="error",
label="Error",
type="labeled",
color="red",
icon="x-circle"
)
]
# Styling
use_custom_header = True
header_class = "bg-gradient-to-r from-blue-500 to-purple-500"
body_class = "bg-gray-50"
# Validation
shadow_on_error = "xl"
errors_at = "bottom"
# Fields
class FieldsModel(BaseModel):
threshold: float = Field(0.5, ge=0.0, le=1.0)
mode: str = Field("auto", pattern="^(auto|manual)$")
# Required method
@classmethod
def get_default_values(cls):
return cls.FieldsModel().model_dump()
# Optional methods
@classmethod
def process(cls, inputs: dict, field_values: dict):
"""Process input data."""
data = inputs.get("input")
threshold = field_values["threshold"]
if data is None:
return {"error": "No input data"}
try:
result = data * threshold
return {"output": result}
except Exception as e:
return {"error": str(e)}
@classmethod
def validate_inputs(cls, inputs: dict):
"""Validate inputs before processing."""
return "input" in inputs and inputs["input"] is not None
@classmethod
def render_custom_header(cls, field_values: dict):
"""Generate custom header HTML."""
mode = field_values.get("mode", "auto")
return f'<div class="text-white font-bold">Mode: {mode.upper()}</div>'
Registration¶
from pynodewidget import NodeFlowWidget
flow = NodeFlowWidget()
# Register nodes
flow.register_node(MinimalNode)
flow.register_node(FullNode)
# Nodes are now available in the sidebar
Type Checking¶
Use protocols for type safety:
from typing import TYPE_CHECKING
if TYPE_CHECKING:
from pynodeflow.protocols import NodeFactory
def register_custom_node(node_class: type[NodeFactory]):
"""Type-safe node registration."""
# node_class is guaranteed to have required attributes
print(f"Registering {node_class.label}")
# Type checker verifies protocol compliance
register_custom_node(MinimalNode) # ✓ OK
register_custom_node(FullNode) # ✓ OK
register_custom_node(dict) # ✗ Type error
Advanced Usage¶
Dynamic Handle Specs¶
Generate handles programmatically:
from pynodeflow.protocols import HandleSpec
class DynamicNode:
type = "dynamic"
category = "advanced"
label = "Dynamic Node"
@classmethod
def get_input_specs(cls, num_inputs: int):
"""Generate input specs dynamically."""
return [
HandleSpec(
name=f"input_{i}",
label=f"Input {i+1}",
type="labeled",
color="blue"
)
for i in range(num_inputs)
]
# Set during initialization
input_specs: list[HandleSpec] = []
class FieldsModel(BaseModel):
num_inputs: int = Field(2, ge=1, le=10)
@classmethod
def get_default_values(cls):
return cls.FieldsModel().model_dump()
# Before registration, configure handles
DynamicNode.input_specs = DynamicNode.get_input_specs(3)
flow.register_node(DynamicNode)
Conditional Handles¶
Show/hide handles based on field values:
class ConditionalNode:
type = "conditional"
category = "advanced"
label = "Conditional Node"
inputs = ["input"]
outputs = ["output"]
# Additional optional output
output_specs = [
HandleSpec(name="output", label="Output"),
HandleSpec(name="debug", label="Debug Output")
]
class FieldsModel(BaseModel):
enable_debug: bool = False
@classmethod
def get_default_values(cls):
return cls.FieldsModel().model_dump()
@classmethod
def process(cls, inputs, field_values):
result = {"output": inputs.get("input")}
if field_values.get("enable_debug"):
result["debug"] = {"raw_input": inputs}
return result
Metadata Inspection¶
from pynodeflow.protocols import NodeMetadata
# Extract metadata
metadata = NodeMetadata.from_class(FullNode)
# Inspect fields
print(f"Node type: {metadata.type}")
print(f"Category: {metadata.category}")
print(f"Has custom header: {metadata.use_custom_header}")
# Access schema
schema = metadata.fields_schema
print(f"Required fields: {schema.get('required', [])}")
# Get defaults
defaults = metadata.default_values
print(f"Default values: {defaults}")
# Serialize for JavaScript
js_data = metadata.model_dump()
Protocol Implementation Checklist¶
Use this checklist when creating custom nodes:
Required ✅:
- [ ] type: str - Unique identifier
- [ ] category: str - Organization category
- [ ] label: str - Display name
- [ ] FieldsModel: Type[BaseModel] - Pydantic model
- [ ] get_default_values() - Returns defaults dict
Recommended 📝:
- [ ] description: str - Tooltip text
- [ ] icon: str - Visual identifier
- [ ] inputs: list[str] or input_specs: list[HandleSpec]
- [ ] outputs: list[str] or output_specs: list[HandleSpec]
Optional 🎨:
- [ ] color: str - Tailwind color class
- [ ] use_custom_header: bool + render_custom_header()
- [ ] use_custom_footer: bool + render_custom_footer()
- [ ] header_class: str - Header styling
- [ ] body_class: str - Body styling
- [ ] footer_class: str - Footer styling
- [ ] shadow_on_error: str - Error shadow size
- [ ] errors_at: str - Error position
Functional ⚙️:
- [ ] process() - Computation logic
- [ ] validate_inputs() - Input validation
Examples¶
Data Processing Node¶
from pydantic import BaseModel, Field
from pynodeflow.protocols import HandleSpec
class DataProcessor:
type = "data-processor"
category = "data"
label = "Data Processor"
description = "Process and transform data"
icon = "cpu"
input_specs = [
HandleSpec(name="data", label="Data", type="labeled", color="blue")
]
output_specs = [
HandleSpec(name="result", label="Result", type="labeled", color="green")
]
class FieldsModel(BaseModel):
operation: str = Field("multiply", pattern="^(multiply|divide|add|subtract)$")
factor: float = Field(1.0, description="Operation factor")
@classmethod
def get_default_values(cls):
return cls.FieldsModel().model_dump()
@classmethod
def process(cls, inputs, field_values):
data = inputs.get("data", 0)
operation = field_values["operation"]
factor = field_values["factor"]
if operation == "multiply":
result = data * factor
elif operation == "divide":
result = data / factor if factor != 0 else 0
elif operation == "add":
result = data + factor
else: # subtract
result = data - factor
return {"result": result}
Visualization Node¶
class ChartNode:
type = "chart"
category = "visualization"
label = "Chart"
description = "Display data as chart"
icon = "bar-chart"
color = "purple"
inputs = ["data"]
use_custom_footer = True
footer_class = "bg-purple-100 p-2"
class FieldsModel(BaseModel):
chart_type: str = Field("bar", pattern="^(bar|line|pie)$")
title: str = "Data Visualization"
@classmethod
def get_default_values(cls):
return cls.FieldsModel().model_dump()
@classmethod
def render_custom_footer(cls, field_values):
title = field_values.get("title", "Chart")
chart_type = field_values.get("chart_type", "bar")
return f'<div class="text-sm">📊 {title} ({chart_type})</div>'
See Also¶
- Creating Custom Nodes: Guide on building nodes
- JsonSchemaNodeWidget: Base class implementing protocol
- Handle Configuration: Working with HandleSpec