Recipe: Supabase Python writeback

{
  "type": "writeback",
  "writeback": {
    "kind": "python",
    "python": {
      "code": "…see below…",
      "secrets": [{ "key": "SUPABASE_DB_URL", "env": "SUPABASE_DB_URL" }]
    }
  },
  "inputs": ["clean_vehicles"]
}

import os
import sqlalchemy as sa
from _panels import input_table, writeback

# Bulk input: the node's single table input, staged for this invocation.
df = input_table().df()

engine = sa.create_engine(os.environ["SUPABASE_DB_URL"])

writeback.record_request({
    "destination": "supabase",
    "operation": "upsert_vehicles_fn",
    "rows": len(df),
})

# Bespoke mutation: your own transaction, e.g. a Postgres function that
# applies domain-specific conflict rules the native upsert can't express.
with engine.begin() as conn:
    for chunk_start in range(0, len(df), 500):
        chunk = df.iloc[chunk_start:chunk_start + 500]
        conn.execute(
            sa.text("SELECT app.upsert_vehicles(CAST(:payload AS jsonb))"),
            {"payload": chunk.to_json(orient="records")},
        )
        writeback.record_progress({"rows_delivered": chunk_start + len(chunk)})

writeback.record_response(200, {"rows_written": len(df), "status": "success"})