Python in Pipelines

Two pipeline nodes run Python: Python Script (once per message) and Batch Transform (once per item, in parallel).

The transform contract

# Supported modules: system, math, json, datetime
# Globals: payload (data), message (metadata)

def transform(payload, message):
    """
    :param payload: the primary data object from the previous node
    :param message: the full message including metadata and context
    :return: the value passed to the next node
    """
    return payload

payload	What the previous node produced
message	The full envelope — payload, metadata, correlation, context
return value	Passed to the next node
return None	Ends this branch — the idiomatic filter

Batch Transform

def transform(item, message):
    """
    :param item: the current item of the incoming array
    :param message: the parent message context
    :return: the transformed item
    """
    return {**item, "celsius": round((item["fahrenheit"] - 32) * 5 / 9, 2)}

Items are processed in parallel and the array stays intact. Prefer this over Array Iterator when items are independent — Array Iterator is for when each item must traverse the rest of the graph separately.

Working with tags

def transform(payload, message):
    tags = system.tag.readAll([
        "Line1/Filler/Motor1/Speed",
        "Line1/Filler/Motor1/Current",
        "Line1/Filler/Motor1/Running",
    ])

    if any(t["quality"] != "Good" for t in tags):
        return None                       # do not compute from untrustworthy data

    speed, current, running = (t["value"] for t in tags)
    return {
        "speed": speed,
        "current": current,
        "running": running,
        "load_pct": round(current / 12.5 * 100, 1) if running else 0.0,
    }

Two habits that matter: check quality, and batch reads. readAll is one round trip; a list comprehension of read is N.

Writing to a database

def transform(payload, message):
    system.db.runNamedQuery("insert_production", {
        "ts":    payload["ts"],
        "line":  payload["line"],
        "units": payload["units"],
    })
    return payload

Prefer routing writes through a SQL Query node rather than doing them in Python: the node's writes are covered by store-and-forward, so an unreachable database buffers and replays instead of failing.

Error handling

An uncaught exception fails the node; the branch stops and the error is logged against the execution. Other branches continue.

Handle it explicitly when you want a different outcome:

def transform(payload, message):
    try:
        rows = system.db.runNamedQuery("lookup_recipe", {"sku": payload["sku"]})
    except Exception as e:
        return {"error": str(e), "sku": payload.get("sku")}   # route to a notify branch

    if not rows:
        return None                                          # unknown SKU: drop quietly
    return {**payload, "recipe": rows[0]}

Capability scopes

Script calls are gated by capability scopes, enforced at the worker. A script cannot reach an API its execution context is not scoped for.

Call	Required scope
system.tag.read	tag:read
system.tag.write	tag:write
system.db.runNamedQuery	db:read
system.db.runQuery, callProc, any transaction call	db:write
system.hist.getTrends	hist:read
system.vars.get / keys / list	vars:read
system.vars.set / delete / incr	vars:write
system.files.read	file:read
system.files.save	file:write
system.message.send	script:execute

Note that runQuery requires db:write while runNamedQuery requires only db:read — arbitrary SQL may write, a named query cannot. That is a concrete reason to prefer named queries: they let you run scripts at a lower privilege.

The gate is fail-safe: an unmapped method on a sensitive namespace falls back to that namespace's strong scope, so a newly-added API is gated by default rather than silently open.

Timeouts

The Python node's default timeout is 60 seconds. Exceeding it fails the node.

If work legitimately takes minutes, it does not belong in a pipeline node driven by a fast trigger. Move it to a Gateway script, or split it across a queue.

Performance

Symptom	Fix
High CPU on the Python worker	Filter with RBE before the Python node; not every message needs Python
Slow tag access	One readAll instead of a loop of read
Slow database access	One query returning many rows, not many queries returning one
Memory growth	Stream with Array Iterator instead of materialising large arrays
Pipeline falls behind its trigger	Lengthen the interval, or split the work across pipelines

Set-based work belongs in SQL. Python is for the logic SQL cannot express, not for iterating rows.

Development workflow

1. Write the transform in the node editor; the linter flags syntax errors as you type.
2. Attach a Debug node downstream and watch the output.
3. Use trigger-node to run just this node with its current input.
4. Read the execution log for failures.

→ Execution & debugging

Next

→ Gateway scripts