Tutorial: Log Data to SQL

Build a pipeline that reads tags on a schedule, computes a summary, and writes it to a SQL database — with store-and-forward so an unavailable database does not lose data.

You will need: tags with live data, and a SQL database you can write to. Time: ~30 minutes.

---

Step 1 — Create the connection

Connections → New → SQL database.

Field	Value
Name	ReportingDB
Driver	postgresql
Host / Port / Database	Your server
Username / Password	A dedicated account with rights only on the target table
Max connections	10

Test, and save.

→ SQL databases

---

Step 2 — Create the target table

CREATE TABLE production_log (
  id          BIGSERIAL PRIMARY KEY,
  ts          TIMESTAMPTZ NOT NULL,
  line        TEXT        NOT NULL,
  avg_speed   DOUBLE PRECISION,
  avg_current DOUBLE PRECISION,
  run_minutes DOUBLE PRECISION,
  UNIQUE (ts, line)
);

CREATE INDEX ON production_log (line, ts DESC);

The unique constraint on (ts, line) is deliberate: store-and-forward may replay a batch that was written but not acknowledged. With the constraint, a replay is rejected as a duplicate instead of double-counting your production figures.

---

Step 3 — Write the insert

The SQL node runs the statement you type into it. Values come from the incoming message through {{…}}, and each one is handed to the driver as a bound parameter rather than pasted into the text — so this is parameterised, not string-building.

INSERT INTO production_log (ts, line, avg_speed, avg_current, run_minutes)
VALUES ({{payload.ts}}, {{payload.line}}, {{payload.avg_speed}},
        {{payload.avg_current}}, {{payload.run_minutes}})
ON CONFLICT (ts, line) DO NOTHING

Keep the statement to hand — it goes into the node in step 6.

ON CONFLICT DO NOTHING makes the write idempotent, which is what you want under replay.

Not the same as a named query

Named queries use :name and belong to screen bindings in the Designer. The pipeline SQL node is a different thing: it takes a raw statement and fills {{…}} from the message.

---

Step 4 — Build the pipeline

Inject (cron * * * * *) → Python (sample and aggregate) → SQL Query (insert)
                                   ↓
                                 Debug

Inject

Setting	Value
Repeat	Cron * * * * * — every minute

The pipeline samples the tags once a minute and writes one row when the hour turns.

---

Step 5 — Aggregate in Python

An instantaneous read is a snapshot, not an hour's average. The Python node keeps a running total for the current hour in a variable, and emits the finished hour's row when the clock moves into the next one:

import datetime

SPEED   = "Site1/Line1/Filler/Motor1/Speed"
CURRENT = "Site1/Line1/Filler/Motor1/Current"
RUNNING = "Site1/Line1/Filler/Motor1/Running"
ACC     = "prod_hour_acc"

def transform(payload, message):
    hour = datetime.datetime.now(datetime.timezone.utc).replace(
               minute=0, second=0, microsecond=0).isoformat()

    acc = system.vars.get(ACC, None)
    row = None
    if acc and acc["hour"] != hour:
        if acc["n"] > 0:                  # the hour has turned: emit the finished one
            row = {
                "ts":          acc["hour"],
                "line":        "L1",
                "avg_speed":   acc["speed"] / acc["n"],
                "avg_current": acc["current"] / acc["n"],
                "run_minutes": float(acc["run"]),
            }
        acc = None
    if not acc:
        acc = {"hour": hour, "n": 0, "speed": 0.0, "current": 0.0, "run": 0}

    readings = {r["tag_path"]: r for r in system.tag.readAll([SPEED, CURRENT, RUNNING])}
    if all(readings.get(p, {}).get("quality") == "Good" for p in (SPEED, CURRENT, RUNNING)):
        acc["n"]       += 1
        acc["speed"]   += readings[SPEED]["value"]
        acc["current"] += readings[CURRENT]["value"]
        acc["run"]     += 1 if readings[RUNNING]["value"] else 0    # one sample = one minute
    system.vars.set(ACC, acc, "persistent")

    return row                            # None ends the branch until the hour turns

Two habits again: check quality before acting — a sample that is not Good is left out of the average rather than counted — and return None to end the branch rather than writing a row you do not trust.

Why not system.hist?

In this release system.hist.getTrends does not work from a Python node: every call returns None. Accumulating the samples in the pipeline, as above, is the approach that works.

---

Step 6 — Write with the SQL node

Setting	Value
Connection	ReportingDB
Query	The INSERT from step 3, with its {{payload.…}} placeholders

The Python node upstream returns exactly the keys the placeholders name — ts, line, avg_speed, avg_current, run_minutes — which is what makes the two halves line up. Rename a key in the Python and the placeholder resolves to nothing.

Use the SQL node rather than system.db in Python. The node's writes are covered by store-and-forward: if the database is unreachable the write is buffered and replayed in order; if it fails permanently it is quarantined for you to inspect. A script write simply fails.

That is the whole reason this is a pipeline and not a script.

---

Step 7 — Test

1. Attach a Debug node to the Python output.
2. Start the pipeline and wait for the top of the next hour — until then the Python node returns None and nothing reaches the Debug panel. The first row covers a partial hour.
3. Confirm the row landed:

SELECT * FROM production_log ORDER BY ts DESC LIMIT 5;

4. Let it run for a couple of hours.

---

Step 8 — Verify the resilience

This is the step people skip, and it is the one worth doing.

1. Stop the database.
2. Let the pipeline run across the top of an hour or two.
3. Check the Status panel — the buffer is holding the writes.
4. Start the database.
5. Confirm the buffered rows drain, in order, with no duplicates.

If duplicates appear, the unique constraint from step 2 is missing.

---

Step 9 — Operate it

Watch	Where
Executions and failures	The execution log panel
Buffer depth and pressure	The Status panel
Quarantined writes	The Status panel's quarantine list

A quarantined batch usually means a schema mismatch, a constraint violation or a credential change. Fix the cause before retrying, or it will re-quarantine.

Remove the Debug node before this goes to production. A Debug node on a busy branch is a real cost.

---

Variations

Incremental extract — track a watermark in a variable so a restart resumes rather than reprocessing:

last = system.vars.get("prod_watermark", "1970-01-01T00:00:00Z")
# ... query changes since `last` ...
system.vars.set("prod_watermark", newest, "persistent")

Event-driven instead of scheduled — replace Inject with an Industrial I/O subscribe node so a batch-complete signal triggers the write.

High-volume time-series — write to QuestDB with the QuestDB node in ingress mode instead of SQL.

Transactional multi-table write — move it to a Gateway script so all tables commit or none do.

Next

→ Publish a REST endpoint · Pipelines