Sample — High Upvotes Alert #

Trigger when a post’s upvotes exceed a configured threshold.

Workflow structure #

• Input: Post Upvotes
• Primitive: Integer (threshold)
• Logic: Greater Than
• Output: Trigger Notification

Example #

from notifications import WorkflowGraph, node_registry, WorkflowExecutionEngine from notifications.models import Connection

workflow = WorkflowGraph.create(“High Upvotes Alert”)

upvotes = node_registry.create_node_instance(“INPUT_POST_UPVOTES”, {“x”: 100, “y”: 150}) threshold = node_registry.create_node_instance(“PRIMITIVE_INTEGER”, {“x”: 300, “y”: 150}) threshold.parameters[“value”] = 500 gt = node_registry.create_node_instance(“BOOLEAN_GREATER_THAN”, {“x”: 500, “y”: 150}) trigger = node_registry.create_node_instance(“OUTPUT_TRIGGER_NOTIFICATION”, {“x”: 700, “y”: 150})

for n in [upvotes, threshold, gt, trigger]: workflow.add_node(n)

workflow.add_connection(Connection(upvotes.id, “Upvotes”, gt.id, “A”)) workflow.add_connection(Connection(threshold.id, “Value”, gt.id, “B”)) workflow.add_connection(Connection(gt.id, “Result”, trigger.id, “Condition”))

engine = WorkflowExecutionEngine() print(engine.execute(workflow, {“upvotes”: 1000, “comment_count”: 50, “ai_sentiment”: 0.2}))

=== Workflow JSON (sketch)
```json
{
  "name": "High Upvotes Alert",
  "nodes": [
    {"id": "n1", "type": "INPUT_POST_UPVOTES"},
    {"id": "n2", "type": "PRIMITIVE_INTEGER", "parameters": {"value": 500}},
    {"id": "n3", "type": "BOOLEAN_GREATER_THAN"},
    {"id": "n4", "type": "OUTPUT_TRIGGER_NOTIFICATION"}
  ],
  "connections": [
    {"from": ["n1", "Upvotes"], "to": ["n3", "A"]},
    {"from": ["n2", "Value"],   "to": ["n3", "B"]},
    {"from": ["n3", "Result"],  "to": ["n4", "Condition"]}
  ]
}

> [!TIP]
> Add a `Comment Count` condition and combine with `AND` to reduce false positives.