Use a temporary database for one reviewed calculation
An in-memory SQLite database lets a short Python task use SQL filters and grouping without creating a database file. This recipe validates four synthetic orders, inserts them with bound parameters and calculates revenue for paid orders only. Pending orders remain in the input count and pass the same validation.
The input object contains only orders, a list of 1–100 records within 16 KiB of UTF-8 JSON. Each record has exactly the five fields below. Prices are EUR integer cents by this recipe's contract. Missing fields, duplicate IDs and invalid pending orders reject the entire input before a database is opened.
Scroll horizontally to see every column.
| Field | Accepted value |
|---|---|
| order_id | Unique ORD- followed by exactly three ASCII digits |
| channel | 1–48 characters; no surrounding whitespace, control characters or unpaired surrogates |
| units | Integer from 1 to 1000; booleans and numeric strings are rejected |
| unit_cents | Integer from 0 to 1000000, interpreted as EUR cents |
| status | Exactly paid or pending |
Keep order identity separate from payment status
{ "orders": [ { "order_id": "ORD-201", "channel": "direct", "units": 2, "unit_cents": 1200, "status": "paid" }, { "order_id": "ORD-202", "channel": "partner", "units": 3, "unit_cents": 800, "status": "paid" }, { "order_id": "ORD-203", "channel": "direct", "units": 1, "unit_cents": 2400, "status": "paid" }, { "order_id": "ORD-204", "channel": "partner", "units": 1, "unit_cents": 1500, "status": "pending" } ]}The paid direct orders contribute 2,400 cents each. The paid partner order contributes another 2,400. A pending partner order worth 1,500 cents is excluded from recognized revenue. The expected paid total is 7,200 cents, six units and three orders, from four input records.
Duplicate object keys and noninteger JSON numbers are rejected before field checks. That includes NaN, Infinity, 1.0 and exponent notation. A boolean is also rejected as a numeric field, even though Python's bool is a subclass of int. These are explicit business rules rather than type coercion. Python JSON decoder options.
Bind values and keep the SQL fixed
from contextlib import closingimport jsonimport reimport sqlite3import sysimport unicodedataINPUT_JSON = "{\"orders\":[{\"order_id\":\"ORD-201\",\"channel\":\"direct\",\"units\":2,\"unit_cents\":1200,\"status\":\"paid\"},{\"order_id\":\"ORD-202\",\"channel\":\"partner\",\"units\":3,\"unit_cents\":800,\"status\":\"paid\"},{\"order_id\":\"ORD-203\",\"channel\":\"direct\",\"units\":1,\"unit_cents\":2400,\"status\":\"paid\"},{\"order_id\":\"ORD-204\",\"channel\":\"partner\",\"units\":1,\"unit_cents\":1500,\"status\":\"pending\"}]}"FIELDS = {"order_id", "channel", "units", "unit_cents", "status"}def unique_object(pairs): result = {} for key, value in pairs: if key in result: raise ValueError("Duplicate JSON key.") result[key] = value return resultdef reject_number(value): raise ValueError("Only integer JSON numbers are accepted.")def analyze(text): if not isinstance(text, str) or not text or len(text.encode("utf-8")) > 16 * 1024: raise ValueError("Input must be JSON text within 16 KiB.") try: data = json.loads(text, object_pairs_hook=unique_object, parse_constant=reject_number, parse_float=reject_number) except (ValueError, RecursionError): raise ValueError("Invalid JSON, duplicate key or noninteger number.") from None if not isinstance(data, dict) or set(data) != {"orders"}: raise ValueError("Expected an object containing only orders.") orders = data["orders"] if not isinstance(orders, list) or not 1 <= len(orders) <= 100: raise ValueError("Expected 1 to 100 orders.") rows = [] seen = set() for index, order in enumerate(orders, 1): prefix = "Order " + str(index) + ": " if not isinstance(order, dict) or set(order) != FIELDS: raise ValueError(prefix + "fields do not match the contract.") order_id = order["order_id"] if not isinstance(order_id, str) or not re.fullmatch(r"ORD-[0-9]{3}", order_id) or order_id in seen: raise ValueError(prefix + "invalid or duplicate ID.") channel = order["channel"] if not isinstance(channel, str) or not 1 <= len(channel) <= 48 or channel != channel.strip(): raise ValueError(prefix + "invalid channel.") if any(unicodedata.category(c) in {"Cc", "Cs"} for c in channel): raise ValueError(prefix + "invalid channel characters.") if type(order["units"]) is not int or not 1 <= order["units"] <= 1000: raise ValueError(prefix + "units must be an integer from 1 to 1000.") if type(order["unit_cents"]) is not int or not 0 <= order["unit_cents"] <= 1_000_000: raise ValueError(prefix + "unit_cents must be an integer from 0 to 1000000.") if not isinstance(order["status"], str) or order["status"] not in {"paid", "pending"}: raise ValueError(prefix + "status must be paid or pending.") seen.add(order_id) rows.append((order_id, channel, order["units"], order["unit_cents"], order["status"])) # Validate every order before opening the database. with closing(sqlite3.connect(":memory:")) as connection: connection.row_factory = sqlite3.Row with connection: connection.execute("""CREATE TABLE orders ( order_id TEXT PRIMARY KEY, channel TEXT NOT NULL, units INTEGER NOT NULL CHECK (units BETWEEN 1 AND 1000), unit_cents INTEGER NOT NULL CHECK (unit_cents BETWEEN 0 AND 1000000), status TEXT NOT NULL CHECK (status IN ('paid', 'pending')) )""") connection.executemany( "INSERT INTO orders VALUES (?, ?, ?, ?, ?)", rows) summary = connection.execute("""SELECT COUNT(*) AS order_count, COALESCE(SUM(units), 0) AS units, COALESCE(SUM(units * unit_cents), 0) AS revenue_cents FROM orders WHERE status = ?""", ("paid",)).fetchone() channels = connection.execute("""SELECT channel, COUNT(*) AS order_count, SUM(units) AS units, SUM(units * unit_cents) AS revenue_cents FROM orders WHERE status = ? GROUP BY channel ORDER BY channel COLLATE BINARY""", ("paid",)).fetchall() return { "currency": "EUR", "input_count": len(rows), "paid_order_count": summary["order_count"], "units": summary["units"], "revenue_cents": summary["revenue_cents"], "channels": [dict(row) for row in channels], }if __name__ == "__main__": try: print(json.dumps(analyze(INPUT_JSON), ensure_ascii=False, sort_keys=True, allow_nan=False)) except (ValueError, UnicodeError) as error: message = "Input must be UTF-8 text." if isinstance(error, UnicodeError) else str(error) print(message, file=sys.stderr) sys.exit(1) except sqlite3.Error: print("SQLite calculation failed.", file=sys.stderr) sys.exit(1)python3 --versionpython3 -c 'import sqlite3; print(sqlite3.sqlite_version)'python3 sqlite-orders.pysqlite3.connect(":memory:") creates the temporary database. executemany binds each validated row to question-mark placeholders; the filter binds paid as a value. Input never becomes SQL source, a table name or an ORDER BY expression. The fixed GROUP BY query sorts channels with BINARY collation. Python sqlite3 reference, parameter binding.
The connection context commits or rolls back the insertion transaction. It does not close the connection; contextlib.closing does that on success and failure. Nothing survives the connection closing. SQL errors have a fixed diagnostic, and invalid data produces an order number and failed rule without echoing row values. Connection context manager.
Verify the complete summary, including excluded orders
{ "currency": "EUR", "input_count": 4, "paid_order_count": 3, "units": 6, "revenue_cents": 7200, "channels": [ { "channel": "direct", "order_count": 2, "units": 3, "revenue_cents": 4800 }, { "channel": "partner", "order_count": 1, "units": 3, "revenue_cents": 2400 } ]}Each channel total comes from paid rows. Channels containing only pending orders are absent. If all orders are pending, paid_order_count, units and revenue_cents are zero and channels is empty. Input_count still records every validated order. The chosen bounds also keep the largest possible sum below SQLite's signed 64-bit integer limit.
import assert from "node:assert/strict"import { readFileSync } from "node:fs"const expected = { "currency": "EUR", "input_count": 4, "paid_order_count": 3, "units": 6, "revenue_cents": 7200, "channels": [ { "channel": "direct", "order_count": 2, "units": 3, "revenue_cents": 4800 }, { "channel": "partner", "order_count": 1, "units": 3, "revenue_cents": 2400 } ]}try { const actual = JSON.parse(readFileSync("result.json", "utf8")) assert.deepStrictEqual(actual, expected) console.log("Fixture result matches.")} catch { console.error("Fixture result did not match the expected contract.") process.exitCode = 1}python3 sqlite-orders.py > result.json && node verify-result.mjsThis verifier checks exact fixture equality. When adapting the dataset, define your own result contract and check that channel counts, units and revenue reconcile with the overall paid totals. Keep input_count distinct from paid_order_count.
Submit the same source to a Python worker
Connect an online Docker worker and create a workspace API key through the quickstart. Save the shared client from JavaScript code execution as run-job.mjs beside sqlite-orders.py and verify-result.mjs. The local client reads the program and sends source text with the python template. It does not upload a database or JSON file.
# CPUOS_API_KEY is already set in the trusted submitting environment.# Generate once for this intended execution; retain it for recovery.export CPUOS_IDEMPOTENCY_KEY="$(node -p 'crypto.randomUUID()')"node run-job.mjs python sqlite-orders.py > result.json && node verify-result.mjsThe client requests 1 CPU, 256 MiB and a 30-second timeout. It accepts parsed JSON only after completed status, exit code 0, no execution error and untruncated output. Reuse the same idempotency key and unchanged source after an uncertain submission. See Python jobs API for lifecycle handling and structured results for application checks.
cpuOS runs trusted team code on your restricted Docker worker. The source and output pass through the EU-hosted control plane; the worker location is your choice. Jobs have no network, package installation, file-transfer API or persistent database. Keep credentials outside job source and the full submitted program within 64 KiB.
Change queries in reviewed code, not in incoming data
- Test a pending-only batch, a duplicate order ID and a boolean units value before changing the contract.
- Keep query text and identifiers fixed. Parameter placeholders bind values, not SQL structure.
- Add an explicit contract before combining currencies, handling refunds or recognizing revenue at another payment stage.
- Use this temporary database for a small calculation. Durable application records need a separate storage design.
Use the recipe as an approved n8n workflow step or LangChain tool. For an export in CSV form, start with the CSV parser. Model-generated SQL needs separate review; this recipe accepts order data and executes only its fixed statements.